fix: require redrive reuse authority for replay time

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,21 @@ 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
+    # 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
+    )
+
+
 class BusinessReplayClock:
     """Index identities once, then read time only at actual scalar consumption."""
 
@@ -133,6 +149,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 +235,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:

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"
+        }
+      }
+    ]
+  }
+}

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",

tests/test_business_replay_clock_redrive.py

@@ -0,0 +1,142 @@
+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)
+
+
+@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]
+    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