From b6dc2db48a2337df467364a18feb0c77c2bdc0fd Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:22:10 +0000 Subject: [PATCH 1/2] fix: require redrive reuse authority before exposing replay time --- src/durable_workflow/replay_clock.py | 34 ++++- ...ess-clock-redrive-timer-missing-reuse.json | 92 ++++++++++++ tests/redrive_fixture_registry.py | 11 ++ tests/test_business_replay_clock_redrive.py | 131 ++++++++++++++++++ 4 files changed, 264 insertions(+), 4 deletions(-) create mode 100644 tests/fixtures/replay_regressions/business-clock-redrive-timer-missing-reuse.json create mode 100644 tests/test_business_replay_clock_redrive.py diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index e827081..6334e28 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -26,6 +26,7 @@ _CHILD_IDENTITIES = ("child_call_id", "child_workflow_instance_id", "child_workflow_run_id", "child_workflow_type") _CHILD_EVENTS = frozenset({"ChildWorkflowScheduled", "ChildRunStarted", *_CHILD_TERMINALS}) _CHILD_FAMILY = "child workflow" +_REUSE_FIELDS = ("reused_from_run_id", "reused_recorded_at", "reused_activity_execution_id") def _task() -> object | None: @@ -123,6 +124,18 @@ def _identity(value: Any) -> TypeGuard[str]: return isinstance(value, str) and bool(value.strip()) +def _declares_redrive(event: Mapping[str, Any]) -> bool: + payload = event.get("payload") + if not isinstance(payload, Mapping): + return False + if _kind(event) == "StartAccepted": + return payload.get("outcome") == "redriven" + return _kind(event) == "WorkflowStarted" and ( + payload.get("recovery_kind") == "redrive" + or "replayed_started_at" in payload or "resume_step_sequence" in payload + ) + + class BusinessReplayClock: """Index identities once, then read time only at actual scalar consumption.""" @@ -133,6 +146,7 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id self._source_run: str | None = None self._resume_step: int | None = None self._redrive_inserted: datetime | None = None + self._redrive_declared = any(_declares_redrive(event) for event in events) starts = [index for index, event in enumerate(events) if _kind(event) == "WorkflowStarted"] self.initial: tuple[int | None, datetime | None] = (None, None) if len(starts) == 1: @@ -218,12 +232,24 @@ def observe(self, index: int | None, family: str, sequence: int) -> None: return event = self.events[index] payload = event["payload"] - if _kind(event) in {"ActivityCompleted", "TimerFired"} and "reused_recorded_at" in payload: - time = self._reused_time(payload, family, sequence) - else: - time = _time(_recorded_value(event)) + time = (_time(_recorded_value(event)) if family == _CHILD_FAMILY + else self._scalar_time(event, payload, family, sequence)) self._pending = (index, time) + def _scalar_time( + self, event: Mapping[str, Any], payload: Mapping[str, Any], family: str, sequence: int, + ) -> datetime | None: + if self._redrive_declared and self._source_run is None: + return None + before_resume = self._resume_step is not None and sequence < self._resume_step + if _kind(event) in {"ActivityCompleted", "TimerFired"} and ( + before_resume or any(key in payload for key in _REUSE_FIELDS) + ): + # The validated resume boundary implies reuse even when both + # markers are missing. Insertion time cannot replace its source. + return self._reused_time(payload, family, sequence) + return _time(_recorded_value(event)) + def _child_boundary(self, index: int, sequence: int) -> bool: indices = self._by_sequence.get(sequence, []) if len(indices) not in (2, 3) or indices[-1] != index: diff --git a/tests/fixtures/replay_regressions/business-clock-redrive-timer-missing-reuse.json b/tests/fixtures/replay_regressions/business-clock-redrive-timer-missing-reuse.json new file mode 100644 index 0000000..517d782 --- /dev/null +++ b/tests/fixtures/replay_regressions/business-clock-redrive-timer-missing-reuse.json @@ -0,0 +1,92 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "business-clock-redrive-timer-missing-reuse", + "protocol_version": "1.20", + "bindings": [ + "python" + ], + "workflow": { + "type": "tests.replay.business-clock-core-prefix-timer", + "input": [ + "Taylor", + 5 + ], + "payload_codec": "avro", + "workflow_id": "regression-corpus-redrive-successor-timer", + "run_id": "01JBREDRIVESUCCESSORRUN0000" + }, + "history": [ + { + "sequence": 1, + "event_type": "WorkflowStarted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "workflow_type": "tests.replay.business-clock-core-prefix-timer", + "workflow_instance_id": "regression-corpus-redrive-successor-timer", + "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000", + "workflow_command_id": "01JBREDRIVECOMMAND00000000", + "continued_from_run_id": "01JBREDRIVESOURCERUN000000", + "resume_step_sequence": 3, + "recovery_kind": "redrive", + "execution_deadline_at": null, + "run_deadline_at": null, + "replayed_started_at": "2026-11-03T00:00:00+00:00", + "workflow_link_id": "01JBREDRIVELINK00000000000" + } + }, + { + "sequence": 2, + "event_type": "ActivityCompleted", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "activity_execution_id": "01JBREDRIVESUCCESSORACT000", + "activity_type": "tests.redrive.first", + "sequence": 1, + "result": "wwHioz3/VYAiNwocSGVsbG8sIFRheWxvciE=", + "payload_codec": "avro", + "reused_from_run_id": "01JBREDRIVESOURCERUN000000", + "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0", + "reused_recorded_at": "2026-11-03T00:00:00+00:00" + } + }, + { + "sequence": 3, + "event_type": "TimerScheduled", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z" + } + }, + { + "sequence": 4, + "event_type": "TimerFired", + "recorded_at": "2026-11-03T00:05:00+00:00", + "payload": { + "timer_id": "01JBREDRIVESUCCESSORTIMER0", + "sequence": 2, + "delay_seconds": 5, + "fire_at": "2026-11-03T00:00:05.000000Z", + "fired_at": "2026-11-03T00:00:06.000000Z" + } + } + ], + "expected": { + "command_sequence": [ + { + "type": "complete_workflow", + "result": { + "clock": [ + "2026-11-03T00:00:00.000000+00:00", + "2026-11-03T00:00:00.000000+00:00", + null + ], + "legacy": "2026-11-03T00:00:00.000000+00:00" + } + } + ] + } +} diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py index e349f7b..ac87a23 100644 --- a/tests/redrive_fixture_registry.py +++ b/tests/redrive_fixture_registry.py @@ -58,6 +58,17 @@ class NonQualifying: } NON_QUALIFYING: dict[str, NonQualifying] = { + "business-clock-redrive-timer-missing-reuse": NonQualifying( + sha256="641be3fcc1a639335c4f2b2c9158ee5972946140b8f1244caec9a3e50c65c7ba", + error="resume_step_sequence must follow the last reused step", + replacement="redrive-successor-reused-timer", + release="unreleased (#215 business replay clock)", + reason=( + "The timer precedes the declared resume step but lacks both reuse fields. Its legacy value still " + "replays, while replay_now must remain unavailable instead of borrowing successor insertion time. " + "This is malformed-input refusal evidence, not a history the Core writer can produce." + ), + ), "redrive-start-replayed-clock": NonQualifying( sha256="70653171faf5ddba5ed264d0a7ce8bb1af97112dbd6e3bb0e39c6f155fee0fd2", error="history order", diff --git a/tests/test_business_replay_clock_redrive.py b/tests/test_business_replay_clock_redrive.py new file mode 100644 index 0000000..7e6f344 --- /dev/null +++ b/tests/test_business_replay_clock_redrive.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +from typing import Any + +import pytest + +from durable_workflow.errors import NonDeterministicReplayError +from durable_workflow.workflow import CompleteWorkflow, replay +from tests.business_clock_workflows import PrefixedCoreTimerBusinessClockWorkflow +from tests.test_replay_clock import redrive_fixture + +ORIGINAL = "2026-11-03T00:00:06.000000+00:00" +INSERTED = "2026-11-03T00:05:00.000000+00:00" +REUSE_FIELDS = ("reused_from_run_id", "reused_recorded_at") + + +def fixture(prefixed: bool) -> dict[str, Any]: + document = redrive_fixture("business-clock-core-redrive-acceptance-timer") + if not prefixed: + document["history"].pop(0) + for index, event in enumerate(document["history"], start=1): + event["sequence"] = index + return document + + +def result(document: dict[str, Any]) -> dict[str, Any]: + task = document["workflow"] + outcome = replay(PrefixedCoreTimerBusinessClockWorkflow, document["history"], task["input"], + workflow_id=task["workflow_id"], run_id=task["run_id"], payload_codec="avro") + try: + assert len(outcome.commands) == 1 + command = outcome.commands[0] + assert isinstance(command, CompleteWorkflow), command + return dict(command.result) + finally: + outcome.close() + + +@pytest.mark.parametrize("prefixed", [False, True]) +@pytest.mark.parametrize("missing", [REUSE_FIELDS, REUSE_FIELDS[:1], REUSE_FIELDS[1:]]) +def test_prefix_before_resume_requires_both_reuse_fields_without_insertion_time_fallback( + prefixed: bool, missing: tuple[str, ...], +) -> None: + document = fixture(prefixed) + expected = result(document) + assert expected["clock"][-1] == ORIGINAL + for key in missing: + document["history"][-1]["payload"].pop(key) + expected["clock"][-1] = None + assert result(document) == expected + assert result(document) == expected # Cold reconstruction repeats the refusal. + + +@pytest.mark.parametrize("prefixed", [False, True]) +@pytest.mark.parametrize("missing", ["workflow_command_id", "continued_from_run_id", "workflow_link_id", + "workflow_run_id", "resume_step_sequence", "recovery_kind", "replayed_started_at"]) +def test_unbound_redrive_header_cannot_turn_an_unmarked_timer_into_an_ordinary_result( + prefixed: bool, missing: str, +) -> None: + document = fixture(prefixed) + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") + started["payload"].pop(missing) + for key in REUSE_FIELDS: + document["history"][-1]["payload"].pop(key) + assert result(document)["clock"] == [None] * 3 + + +@pytest.mark.parametrize("prefixed", [False, True]) +def test_bound_post_resume_timer_without_reuse_fields_uses_its_own_consumed_time(prefixed: bool) -> None: + document = fixture(prefixed) + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") + started["payload"]["resume_step_sequence"] = 2 + for key in REUSE_FIELDS: + document["history"][-1]["payload"].pop(key) + assert result(document)["clock"][-1] == INSERTED + + +@pytest.mark.parametrize("prefix_fault", ["outcome", "identity"]) +def test_unbound_acceptance_prefix_does_not_lend_timer_insertion_time(prefix_fault: str) -> None: + document = fixture(True) + for key in REUSE_FIELDS: + document["history"][-1]["payload"].pop(key) + if prefix_fault == "outcome": + document["history"][0]["payload"]["outcome"] = "started" + else: + document["history"][0]["payload"]["workflow_run_id"] = "foreign" + assert result(document)["clock"] == [None] * 3 + + +def test_missing_start_retains_the_original_legacy_clock_error() -> None: + document = fixture(True) + document["history"].pop(1) + with pytest.raises(NonDeterministicReplayError, match="missing WorkflowStarted persisted start time"): + result(document) + + +def test_pre_resume_activity_with_a_schedule_cannot_use_insertion_time_when_reuse_fields_are_missing() -> None: + document = fixture(False) + terminal = document["history"][1] + for key in (*REUSE_FIELDS, "reused_activity_execution_id"): + terminal["payload"].pop(key) + document["history"].insert(1, { + "event_type": "ActivityScheduled", "recorded_at": terminal["recorded_at"], + "payload": {key: terminal["payload"][key] for key in ("sequence", "activity_type", "activity_execution_id")}, + }) + for index, event in enumerate(document["history"], start=1): + event["sequence"] = index + assert result(document)["clock"] == ["2026-11-03T00:00:00.000000+00:00", None, ORIGINAL] + + +@pytest.mark.parametrize("continued", [False, True]) +def test_ordinary_and_continued_histories_without_redrive_metadata_keep_their_own_times(continued: bool) -> None: + from tests.test_business_replay_clock import history + from tests.test_business_replay_clock import result as scalar_result + + events = history() + if continued: + events[0]["payload"]["continued_from_run_id"] = "previous-run" + assert scalar_result(events)["clock"] == [ + None, "2026-11-01T00:00:03.123456+00:00", "2026-11-01T00:00:05.654321+00:00", + ] + + +@pytest.mark.parametrize("marker", ["reused_from_run_id", "reused_recorded_at", "reused_activity_execution_id"]) +def test_ordinary_success_with_a_stray_reuse_marker_has_no_proved_source_time(marker: str) -> None: + from tests.test_business_replay_clock import history + from tests.test_business_replay_clock import result as scalar_result + + events = history() + events[2]["payload"][marker] = "unbound" + assert scalar_result(events)["clock"][1] is None From 0fff23d88e13061726042a9ea568c67ed866f7ae Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 6 Nov 2026 05:27:49 +0000 Subject: [PATCH 2/2] fix: retain redrive declaration across conflicting kind aliases --- src/durable_workflow/replay_clock.py | 9 ++++++--- tests/test_business_replay_clock_redrive.py | 11 +++++++++++ 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py index 6334e28..7084eb9 100644 --- a/src/durable_workflow/replay_clock.py +++ b/src/durable_workflow/replay_clock.py @@ -128,9 +128,12 @@ def _declares_redrive(event: Mapping[str, Any]) -> bool: payload = event.get("payload") if not isinstance(payload, Mapping): return False - if _kind(event) == "StartAccepted": - return payload.get("outcome") == "redriven" - return _kind(event) == "WorkflowStarted" and ( + # A conflicting kind alias invalidates authority; it must not erase a + # recognizable redrive envelope and turn it into ordinary history. + aliases = (event.get("event_type"), event.get("type")) + if "StartAccepted" in aliases and payload.get("outcome") == "redriven": + return True + return "WorkflowStarted" in aliases and ( payload.get("recovery_kind") == "redrive" or "replayed_started_at" in payload or "resume_step_sequence" in payload ) diff --git a/tests/test_business_replay_clock_redrive.py b/tests/test_business_replay_clock_redrive.py index 7e6f344..1da78a4 100644 --- a/tests/test_business_replay_clock_redrive.py +++ b/tests/test_business_replay_clock_redrive.py @@ -94,6 +94,17 @@ def test_missing_start_retains_the_original_legacy_clock_error() -> None: result(document) +@pytest.mark.parametrize("prefixed", [False, True]) +@pytest.mark.parametrize("alias", ["WorkflowTaskCompleted", "StartAccepted"]) +def test_conflicting_start_kind_alias_cannot_hide_an_unbound_redrive_header(prefixed: bool, alias: str) -> None: + document = fixture(prefixed) + started = next(event for event in document["history"] if event["event_type"] == "WorkflowStarted") + started["type"] = alias + for key in REUSE_FIELDS: + document["history"][-1]["payload"].pop(key) + assert result(document) == {"clock": [None] * 3, "legacy": "2026-11-03T00:00:00.000000+00:00"} + + def test_pre_resume_activity_with_a_schedule_cannot_use_insertion_time_when_reuse_fields_are_missing() -> None: document = fixture(False) terminal = document["history"][1]