From 819b96b2c6ae5e2363791336933322a63349308c Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 02:56:41 +0000
Subject: [PATCH 1/7] feat: add a consumed scalar business replay clock

---
 src/durable_workflow/__init__.py              |   2 +
 src/durable_workflow/errors.py                |   7 +
 src/durable_workflow/replay_clock.py          | 230 +++++++++++++++
 src/durable_workflow/workflow.py              |  78 +++--
 tests/business_clock_workflows.py             |  60 ++++
 ...usiness-replay-clock-consumed-scalars.json |  98 +++++++
 tests/test_business_replay_clock.py           | 272 ++++++++++++++++++
 tests/test_replay_regression_corpus.py        |   2 +
 8 files changed, 730 insertions(+), 19 deletions(-)
 create mode 100644 src/durable_workflow/replay_clock.py
 create mode 100644 tests/business_clock_workflows.py
 create mode 100644 tests/fixtures/replay_regressions/business-replay-clock-consumed-scalars.json
 create mode 100644 tests/test_business_replay_clock.py

diff --git a/src/durable_workflow/__init__.py b/src/durable_workflow/__init__.py
index e2d9ed81..941e659a 100644
--- a/src/durable_workflow/__init__.py
+++ b/src/durable_workflow/__init__.py
@@ -93,6 +93,7 @@
     NonRetryableError,
     QueryFailed,
     QueryTaskAuthorityLost,
+    ReplayClockUnavailable,
     RuntimeCapabilityUnsupported,
     RuntimeDiscoveryUnavailable,
     SagaCompensationFailed,
@@ -389,6 +390,7 @@
     "NexusOperationResult",
     "NexusServiceCall",
     "NonDeterministicReplayError",
+    "ReplayClockUnavailable",
     "DuplicateRegistrationError",
     "InvalidWorkflowDefinition",
     "NoopMetrics",
diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py
index e5b01b60..f53d0338 100644
--- a/src/durable_workflow/errors.py
+++ b/src/durable_workflow/errors.py
@@ -27,6 +27,13 @@ class DurableWorkflowError(Exception):
     """Base class for every exception raised by the SDK."""
 
 
+class ReplayClockUnavailable(DurableWorkflowError):
+    """The current execution has no proved business replay-time boundary."""
+
+    def __init__(self) -> None:
+        super().__init__("business replay time is unavailable at this execution boundary")
+
+
 class FailureReportTooLarge(DurableWorkflowError, ValueError):
     """Required failure identity or semantics cannot fit the reporting budget.
 
diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
new file mode 100644
index 00000000..0cb50fa3
--- /dev/null
+++ b/src/durable_workflow/replay_clock.py
@@ -0,0 +1,230 @@
+"""Business replay time from supported consumed boundaries, without wall time."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import functools
+import re
+import threading
+from collections.abc import Callable, Iterator, Mapping, Sequence
+from contextvars import ContextVar
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, ParamSpec, TypeVar
+
+from .errors import ReplayClockUnavailable
+
+_P = ParamSpec("_P")
+_T = TypeVar("_T")
+_TIMESTAMP = re.compile(
+    r"[0-9]{4}-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]"
+    r"(?:\.[0-9]{1,6})?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])",
+)
+_ACTIVITY_TERMINALS = frozenset({"ActivityCompleted", "ActivityFailed", "ActivityTimedOut", "ActivityCancelled"})
+
+
+def _task() -> object | None:
+    try:
+        return asyncio.current_task()
+    except RuntimeError:
+        return None
+
+
+@dataclass(frozen=True)
+class _Frame:
+    owner: object | None
+    time: datetime | None
+    thread: int
+    task: object | None
+
+
+_ACTIVE: ContextVar[_Frame | None] = ContextVar("business_replay_clock", default=None)
+
+
+@contextlib.contextmanager
+def execution_scope(owner: object | None, time: datetime | None) -> Iterator[None]:
+    token = _ACTIVE.set(_Frame(owner, time, threading.get_ident(), _task()))
+    try:
+        yield
+    finally:
+        _ACTIVE.reset(token)
+
+
+def isolated(function: Callable[_P, _T]) -> Callable[_P, _T]:
+    """Nested reconstruction and its unqualified callbacks cannot borrow time."""
+    @functools.wraps(function)
+    def invoke(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        with execution_scope(None, None):
+            return function(*args, **kwargs)
+    return invoke
+
+
+def read(owner: object) -> datetime:
+    frame = _ACTIVE.get()
+    if (frame is None or frame.owner is not owner or frame.time is None
+        or frame.thread != threading.get_ident() or frame.task is not _task()):
+        raise ReplayClockUnavailable()
+    return frame.time
+
+
+def _time(value: Any) -> datetime | None:
+    if not isinstance(value, str) or _TIMESTAMP.fullmatch(value) is None:
+        return None
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
+    except (ValueError, OverflowError):
+        return None
+
+
+def _kind(event: Mapping[str, Any]) -> str | None:
+    value = event.get("event_type") or event.get("type")
+    if "event_type" in event and "type" in event and event["event_type"] != event["type"]:
+        return None
+    return value if isinstance(value, str) and value else None
+
+
+def _sequence(payload: Mapping[str, Any]) -> int | None:
+    value = payload.get("sequence")
+    if type(value) is not int or value < 1:
+        return None
+    if "workflow_sequence" in payload:
+        alias = payload["workflow_sequence"]
+        if type(alias) is not int or alias != value:
+            return None
+    return value
+
+
+def _identity(value: Any) -> bool:
+    return isinstance(value, str) and bool(value.strip())
+
+
+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:
+        self.events = events
+        self._by_sequence: dict[int, list[int]] = {}
+        self._source_run: str | None = None
+        self._resume_step: int | None = None
+        self._redrive_inserted: datetime | None = None
+        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:
+            index = starts[0]
+            payload = events[index].get("payload")
+            if (index == 0 and isinstance(payload, Mapping)
+                and self._redrive_start(payload, workflow_id, run_id)):
+                self.initial = (index, _time(payload.get("replayed_started_at")))
+        self._pending = self.initial
+        self._current = self.initial
+        for index, event in enumerate(events):
+            payload = event.get("payload")
+            if isinstance(payload, Mapping) and (sequence := _sequence(payload)) is not None:
+                self._by_sequence.setdefault(sequence, []).append(index)
+
+    def _redrive_start(self, payload: Mapping[str, Any], workflow_id: str, run_id: str) -> bool:
+        if payload.get("recovery_kind") != "redrive":
+            # Ordinary entry needs run.started_at, not history insertion time.
+            return False
+        keys = ("workflow_instance_id", "workflow_run_id", "workflow_command_id", "continued_from_run_id",
+                "workflow_link_id")
+        if not all(_identity(payload.get(key)) for key in keys):
+            return False
+        if ((workflow_id and payload["workflow_instance_id"] != workflow_id)
+            or (run_id and payload["workflow_run_id"] != run_id)):
+            return False
+        resume_step = payload.get("resume_step_sequence")
+        if type(resume_step) is not int or resume_step < 1 or _time(payload.get("replayed_started_at")) is None:
+            return False
+        if any(type(event.get("sequence")) is not int or event["sequence"] != index + 1
+               for index, event in enumerate(self.events)):
+            return False
+        inserted = _time(self.events[0].get("recorded_at"))
+        original = _time(payload.get("replayed_started_at"))
+        if inserted is None or original is None or original > inserted:
+            return False
+        self._source_run = payload["continued_from_run_id"]
+        self._resume_step = resume_step
+        self._redrive_inserted = inserted
+        return True
+
+    def resume(self, owner: object, function: Callable[..., _T], *args: Any) -> _T:
+        self._current, self._pending = self._pending, (None, None)
+        with execution_scope(owner, self._current[1]):
+            return function(*args)
+
+    def snapshot(self) -> datetime | None:
+        # Only this final inspected prefix is proved for a query callback.
+        index, time = self._current
+        return time if index is not None and index == len(self.events) - 1 else None
+
+    def unavailable(self) -> None:
+        self._current = self._pending = (None, None)
+
+    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"}:
+            return
+        if not self._scalar_boundary(index, family, sequence):
+            return
+        event = self.events[index]
+        payload = event["payload"]
+        value = event.get("timestamp") if "timestamp" in event else event.get("recorded_at")
+        if _kind(event) in {"ActivityCompleted", "TimerFired"} and "reused_recorded_at" in payload:
+            if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run
+                or self._resume_step is None or sequence >= self._resume_step):
+                return
+            if family == "activity" and not _identity(payload.get("reused_activity_execution_id")):
+                return
+            value = payload["reused_recorded_at"]
+            original = _time(value)
+            if original is None or self._redrive_inserted is None or original > self._redrive_inserted:
+                return
+        self._pending = (index, _time(value))
+
+    def _scalar_boundary(self, index: int, family: str, sequence: int) -> bool:
+        event = self.events[index]
+        payload = event.get("payload")
+        if not isinstance(payload, Mapping) or _sequence(payload) != sequence:
+            return False
+        kinds = _ACTIVITY_TERMINALS if family == "activity" else frozenset({"TimerFired"})
+        if _kind(event) not in kinds:
+            return False
+        indices = self._by_sequence.get(sequence, [])
+        matching = [position for position in indices if _kind(self.events[position]) in kinds]
+        if matching != [index]:
+            return False
+        resource = "activity_execution_id" if family == "activity" else "timer_id"
+        identity = payload.get(resource)
+        if not _identity(identity):
+            return False
+        opening = "ActivityScheduled" if family == "activity" else "TimerScheduled"
+        allowed = {opening, *kinds}
+        if family == "activity":
+            allowed.add("ActivityStarted")
+        previous = -1
+        for position in indices:
+            row = self.events[position]
+            body = row.get("payload")
+            if (not isinstance(body, Mapping) or _kind(row) not in allowed
+                or position > index or position <= previous or body.get(resource) != identity
+                or body.get("execution_mode") not in (None, "remote") or body.get("local_activity") is True
+                or body.get("parallel_group") is not None or body.get("parallel_group_path") not in (None, [])
+                or body.get("selection_group") is not None
+                or body.get("cancellation_scope_id") not in (None, "root")):
+                return False
+            if family == "timer" and (
+                body.get("timer_kind") not in (None, "sleep", "durable_timer")
+                or "condition_wait_id" in body or "signal_wait_id" in body
+            ):
+                return False
+            previous = position
+        openings = [position for position in indices if _kind(self.events[position]) == opening]
+        starts = [position for position in indices if _kind(self.events[position]) == "ActivityStarted"]
+        if len(starts) > 1 or (starts and (not openings or starts[0] < openings[0])):
+            return False
+        reused = "reused_recorded_at" in payload and self._source_run is not None
+        if len(openings) != 1 and not (family == "activity" and reused and indices == [index]):
+            return False
+        return not openings or openings[0] < index
diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py
index 58ae9bd5..118eece0 100644
--- a/src/durable_workflow/workflow.py
+++ b/src/durable_workflow/workflow.py
@@ -35,7 +35,7 @@
 from datetime import datetime, timezone
 from typing import Any, TypeVar, cast
 
-from . import serializer
+from . import replay_clock, serializer
 from ._cancellation_scope import CancellationScopeHistory, scope_identity
 from ._cancellation_scope_history import (
     CancellationScopeDeliveryIntent,
@@ -68,6 +68,7 @@
     NexusOperationFailed,
     NonDeterministicReplayError,
     QueryFailed,
+    ReplayClockUnavailable,
     SagaCompensationFailed,
     TypedValueError,
     WorkflowCancelled,
@@ -2201,6 +2202,7 @@ 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._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
@@ -2854,6 +2856,20 @@ def now(self) -> datetime:
             )
         return self._current_time
 
+    def replay_now(self) -> datetime:
+        """Return the proved UTC instant of the supported consumed boundary.
+
+        This additive clock supports scalar persisted activities/timers and an
+        explicit redrive start. Ordinary entry, inline/local/grouped operations
+        and unproved callbacks remain unavailable. Repeated reads are stable;
+        equal/backward recorded times retain their values and precision. Adopt
+        business-time decisions through normal workflow versioning. ``now()``
+        remains the legacy fixed-start accessor, and timestamps are not order IDs.
+        """
+        if self._cancellation_context is not None or self._cancellation_scope_id != "root":
+            raise ReplayClockUnavailable()
+        return replay_clock.read(self)
+
     def random(self) -> random.Random:
         return self._rng
 
@@ -2905,6 +2921,7 @@ def __init__(self, logger: _ReplayLogger) -> None:
         self.generator: Any = None
         self.awaitable: Any = None
 
+    @replay_clock.isolated
     def close(self) -> None:
         self.logger._set_replaying(True)
         awaitable, self.awaitable = self.awaitable, None
@@ -3085,9 +3102,10 @@ def _workflow_cls(self, workflow_type: str | None) -> type:
 class _ReplayState:
     outcome: ReplayOutcome
     instance: Any
+    context: WorkflowContext
 
 
-def _consume_replay_result(state: _ReplayState, result: Any) -> Any:
+def _consume_replay_result(state: _ReplayState, result: Any, *, business_clock: bool = False) -> Any:
     """Detach a handler value before discarded replay cleanup can mutate it."""
     if inspect.isawaitable(result):
         if state.outcome._lifetime is not None:
@@ -3095,7 +3113,10 @@ def _consume_replay_result(state: _ReplayState, result: Any) -> Any:
 
         async def consume() -> Any:
             try:
-                return deepcopy(await result)
+                owner = state.context if business_clock else None
+                time = state.context._business_clock.snapshot() if business_clock else None
+                with replay_clock.execution_scope(owner, time):
+                    return deepcopy(await result)
             finally:
                 state.outcome.close()
 
@@ -3506,10 +3527,11 @@ def _prepare_query_state(
 
     def invoke() -> Any:
         try:
-            result = _invoke_workflow_handler(
-                handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=True,
-            )
-            return _consume_replay_result(state, result)
+            with replay_clock.execution_scope(state.context, state.context._business_clock.snapshot()):
+                result = _invoke_workflow_handler(
+                    handler, list(args) if args is not None else [], workflow_cls, method_name, allow_async=True,
+                )
+                return _consume_replay_result(state, result, business_clock=True)
         except QueryFailed:
             state.outcome.close()
             raise
@@ -3559,6 +3581,7 @@ def query_state(
         raise QueryFailed(str(exc)) from exc
 
 
+@replay_clock.isolated
 def apply_update(
     workflow_cls: type,
     history_events: Iterable[dict[str, Any]],
@@ -3798,6 +3821,7 @@ def invoke() -> Any:
     return invoke
 
 
+@replay_clock.isolated
 def _invoke_update_validator(
     validator: Callable[..., Any],
     args: list[Any],
@@ -4795,6 +4819,7 @@ def _assert_cancellation_scope_replay_supported(
             )
 
 
+@replay_clock.isolated
 def _replay_state(
     workflow_cls: type,
     history_events: Iterable[dict[str, Any]],
@@ -5114,6 +5139,7 @@ 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)
     lifetime = _ReplayLifetime(ctx.logger)
 
     def _set_replaying(replaying: bool) -> None:
@@ -5176,7 +5202,7 @@ def _state(
             ],
         )
         outcome._lifetime = lifetime
-        return _ReplayState(outcome=outcome, instance=instance)
+        return _ReplayState(outcome=outcome, instance=instance, context=ctx)
 
     resolved_results: list[Any] = []
     recorded_steps: list[_RecordedStep] = []
@@ -5970,6 +5996,7 @@ def _receiver_condition_wait_bindings() -> dict[int, str | None]:
     update_registry: dict[str, str] = getattr(workflow_cls, "__workflow_updates__", {}) or {}
 
     def _apply_receiver(receiver: _PendingReceiver) -> None:
+        ctx._business_clock.unavailable()
         if receiver.kind == "signal":
             if receiver.name == MESSAGE_STREAM_SIGNAL:
                 ctx._accept_message_stream(receiver.args)
@@ -6160,7 +6187,8 @@ def _refresh_replay_logging() -> None:
     # by a successfully bound body is still an actual workflow failure.
     try:
         _refresh_replay_logging()
-        gen = entry(ctx, *start_input)
+        with replay_clock.execution_scope(ctx, ctx._business_clock.initial[1]):
+            gen = entry(ctx, *start_input)
     except NonDeterministicReplayError:
         raise
     except LocalActivityExecutionAborted:
@@ -6179,6 +6207,9 @@ def _refresh_replay_logging() -> None:
 
     lifetime.generator = gen
 
+    def _resume_business(function: Callable[[Any], Any], value: Any) -> Any:
+        return ctx._business_clock.resume(ctx, function, value)
+
     _set_replaying(True)
 
     next_value: Any = None
@@ -7068,6 +7099,7 @@ def _terminal_state(value: Any, *, include_pending: bool) -> _ReplayState:
         return _state(commands + [CompleteWorkflow(result=value)])
 
     def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _ReplayState | None:
+        ctx._business_clock.unavailable()
         nonlocal cancellation_consumed, authored_sequence, wait_yield_count, advanced_cmd
         if ctx._cancellation_scope_id != "root" or _scope_call_membership(command) != "root":
             raise NonDeterministicReplayError(
@@ -7102,7 +7134,7 @@ def _consume_cancellation(command: Any, boundary: CancellationDelivery) -> _Repl
             authored_sequence += 1
         try:
             _refresh_replay_logging()
-            advanced_cmd = gen.throw(WorkflowCancelled(
+            advanced_cmd = _resume_business(gen.throw, WorkflowCancelled(
                 "workflow cancellation was requested", request_id=boundary.request_id,
                 context=ctx._cancellation_context,
             ))
@@ -7178,6 +7210,7 @@ def _advance_cancellation_clock(indexes: Iterable[int | None]) -> None:
             ctx._observe_cancellation_replay_time(events[index])
 
     def _consume_scope_delivery(command: Any, delivery: ScopeBoundary) -> _ReplayState | None:
+        ctx._business_clock.unavailable()
         nonlocal advanced_cmd, wait_yield_count
         boundary = delivery.boundary
         _assert_cancellation_call_matches(command, boundary)
@@ -7220,7 +7253,7 @@ def _consume_scope_delivery(command: Any, delivery: ScopeBoundary) -> _ReplaySta
         _apply_due_receivers()
         try:
             _refresh_replay_logging()
-            advanced_cmd = gen.throw(WorkflowCancelled(
+            advanced_cmd = _resume_business(gen.throw, WorkflowCancelled(
                 "scope cancellation was requested", request_id=active_context.request_id, context=active_context,
             ))
             return None
@@ -7256,7 +7289,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
             else:
                 try:
                     _refresh_replay_logging()
-                    cmd = gen.send(None) if first else gen.send(next_value)
+                    cmd = _resume_business(gen.send, None if first else next_value)
                 except StopIteration as stop:
                     return _terminal_state(stop.value, include_pending=True)
                 first = False
@@ -7555,7 +7588,9 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     if failed is not None or group_typed_failure is not None:
                         try:
                             _refresh_replay_logging()
-                            advanced_cmd = gen.throw(failed if failed is not None else group_typed_failure)
+                            advanced_cmd = _resume_business(
+                                gen.throw, failed if failed is not None else group_typed_failure,
+                            )
                             continue
                         except StopIteration as stop:
                             return _terminal_state(stop.value, include_pending=False)
@@ -7574,7 +7609,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     ])
                     try:
                         _refresh_replay_logging()
-                        advanced_cmd = gen.throw(
+                        advanced_cmd = _resume_business(gen.throw,
                             DurableOperationCancelled(
                                 cmd.selection_group_id,
                                 cmd.key,
@@ -7613,7 +7648,9 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                 if failure is not None or handle_typed_failure is not None:
                     try:
                         _refresh_replay_logging()
-                        advanced_cmd = gen.throw(failure if failure is not None else handle_typed_failure)
+                        advanced_cmd = _resume_business(
+                            gen.throw, failure if failure is not None else handle_typed_failure,
+                        )
                         continue
                     except StopIteration as stop:
                         return _terminal_state(stop.value, include_pending=True)
@@ -7678,7 +7715,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     except NexusOperationFailed as exc:
                         try:
                             _refresh_replay_logging()
-                            advanced_cmd = gen.throw(exc)
+                            advanced_cmd = _resume_business(gen.throw, exc)
                             continue
                         except StopIteration as stop:
                             return _terminal_state(stop.value, include_pending=False)
@@ -7711,7 +7748,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     except Exception as exc:
                         try:
                             _refresh_replay_logging()
-                            advanced_cmd = gen.throw(exc)
+                            advanced_cmd = _resume_business(gen.throw, exc)
                             continue
                         except StopIteration as stop:
                             return _terminal_state(stop.value, include_pending=True)
@@ -7911,6 +7948,9 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                 if result_cursor < len(resolved_results):
                     _assert_next_step_matches(cmd)
                     val = resolved_results[result_cursor]
+                    step = recorded_steps[result_cursor]
+                    if isinstance(cmd, ScheduleActivity | StartTimer):
+                        ctx._business_clock.observe(step.history_index, step.shape, step.workflow_sequence)
                     if not isinstance(cmd, RecordLocalActivity) or prepare_local_activities:
                         _advance_cancellation_clock([recorded_steps[result_cursor].history_index])
                     result_cursor += 1
@@ -7922,7 +7962,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     if isinstance(val, ActivityFailed | ChildWorkflowFailed | TypedValueError):
                         try:
                             _refresh_replay_logging()
-                            advanced_cmd = gen.throw(val)
+                            advanced_cmd = _resume_business(gen.throw, val)
                             continue
                         except StopIteration as stop:
                             return _terminal_state(stop.value, include_pending=False)
@@ -7945,7 +7985,7 @@ def _advance_selection_clock(base: int, size: int, failure: BaseException | None
                     if isinstance(local_result, ActivityFailed):
                         try:
                             _refresh_replay_logging()
-                            advanced_cmd = gen.throw(local_result)
+                            advanced_cmd = _resume_business(gen.throw, local_result)
                             continue
                         except StopIteration as stop:
                             return _terminal_state(stop.value, include_pending=True)
diff --git a/tests/business_clock_workflows.py b/tests/business_clock_workflows.py
new file mode 100644
index 00000000..1e29eb73
--- /dev/null
+++ b/tests/business_clock_workflows.py
@@ -0,0 +1,60 @@
+"""One corpus consumer that records the old API's absence as an observation."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from durable_workflow import workflow
+from durable_workflow.errors import DurableWorkflowError
+from durable_workflow.workflow import WorkflowContext
+
+
+def clock_text(ctx: WorkflowContext) -> str | None:
+    accessor = getattr(ctx, "replay_now", None)
+    if accessor is None:
+        # Allows the identical corpus consumer to measure the pre-accessor base.
+        return None
+    try:
+        return str(accessor().isoformat(timespec="microseconds"))
+    except DurableWorkflowError as error:
+        if type(error).__name__ != "ReplayClockUnavailable":
+            raise
+        return None
+
+
+@workflow.defn(name="tests.replay.business-clock")
+class ScalarBusinessClockWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        self.ctx = ctx
+        observations = [clock_text(ctx)]
+        legacy = [ctx.now().isoformat(timespec="microseconds")]
+        yield ctx.schedule_activity("clock.first", [])
+        observations.append(clock_text(ctx))
+        legacy.append(ctx.now().isoformat(timespec="microseconds"))
+        yield ctx.start_timer(2)
+        observations.append(clock_text(ctx))
+        legacy.append(ctx.now().isoformat(timespec="microseconds"))
+        elapsed = None
+        if observations[1] is not None and observations[2] is not None:
+            before = datetime.fromisoformat(observations[1])
+            elapsed = (datetime.fromisoformat(observations[2]) - before).total_seconds()
+        return {"clock": observations, "legacy": legacy, "elapsed_seconds": elapsed}
+
+    @workflow.query("clock")
+    def clock(self) -> str | None:
+        return clock_text(self.ctx)
+
+    @workflow.query("async_clock")
+    async def async_clock(self) -> Any:
+        import asyncio
+
+        before = clock_text(self.ctx)
+
+        async def child_read() -> str | None:
+            return clock_text(self.ctx)
+
+        child = await asyncio.create_task(child_read())
+        thread = await asyncio.to_thread(clock_text, self.ctx)
+        await asyncio.sleep(0)
+        return before, child, thread, clock_text(self.ctx)
diff --git a/tests/fixtures/replay_regressions/business-replay-clock-consumed-scalars.json b/tests/fixtures/replay_regressions/business-replay-clock-consumed-scalars.json
new file mode 100644
index 00000000..7cb21293
--- /dev/null
+++ b/tests/fixtures/replay_regressions/business-replay-clock-consumed-scalars.json
@@ -0,0 +1,98 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "business-replay-clock-consumed-scalars",
+  "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": "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": 3
+    },
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "clock-timer",
+        "timer_kind": "durable_timer",
+        "delay_seconds": 2
+      },
+      "sequence": 4
+    },
+    {
+      "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": 5
+    },
+    {
+      "event_type": "WorkflowTaskCompleted",
+      "recorded_at": "2026-12-01T00:00:00Z",
+      "payload": {},
+      "sequence": 6
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "result": {
+          "clock": [
+            null,
+            "2026-11-01T00:00:03.123456+00:00",
+            "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": 2.530865
+        }
+      }
+    ]
+  }
+}
diff --git a/tests/test_business_replay_clock.py b/tests/test_business_replay_clock.py
new file mode 100644
index 00000000..b6488cea
--- /dev/null
+++ b/tests/test_business_replay_clock.py
@@ -0,0 +1,272 @@
+from __future__ import annotations
+
+import asyncio
+from copy import deepcopy
+from datetime import datetime, timedelta
+from typing import Any
+
+import pytest
+
+from durable_workflow import ReplayClockUnavailable, serializer, workflow
+from durable_workflow.errors import ActivityFailed, NonDeterministicReplayError, QueryFailed
+from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, _prepare_query_state, query_state, replay
+from tests.business_clock_workflows import ScalarBusinessClockWorkflow, clock_text
+
+
+def history() -> list[dict[str, Any]]:
+    return [
+        {"event_type": "WorkflowStarted", "recorded_at": "2026-11-01T00:00:00.123456Z", "payload": {}},
+        {"event_type": "ActivityScheduled", "payload": {
+            "sequence": 1, "activity_execution_id": "clock-activity", "activity_type": "clock.first",
+        }},
+        {"event_type": "ActivityCompleted", "recorded_at": "2026-11-01T00:00:03.123456Z", "payload": {
+            "sequence": 1, "activity_execution_id": "clock-activity", "activity_type": "clock.first",
+            "result": serializer.envelope("first"),
+        }},
+        {"event_type": "TimerScheduled", "payload": {
+            "sequence": 2, "timer_id": "clock-timer", "timer_kind": "durable_timer", "delay_seconds": 2,
+        }},
+        {"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,
+        }},
+    ]
+
+
+def result(events: list[dict[str, Any]], workflow: type = ScalarBusinessClockWorkflow) -> Any:
+    outcome = replay(workflow, events, [], payload_codec="avro")
+    try:
+        assert len(outcome.commands) == 1
+        command = outcome.commands[0]
+        assert isinstance(command, CompleteWorkflow), command
+        return command.result
+    finally:
+        outcome.close()
+
+
+def test_exact_consumed_times_and_legacy_clock_survive_cold_replay_without_history_maximum() -> None:
+    events = history()
+    events.append({"event_type": "WorkflowTaskCompleted", "recorded_at": "2026-12-01T00:00:00Z", "payload": {}})
+    original = deepcopy(events)
+    first = result(events)
+    assert first == result(events)
+    assert first == {
+        "clock": [None, "2026-11-01T00:00:03.123456+00:00", "2026-11-01T00:00:05.654321+00:00"],
+        "legacy": ["2026-11-01T00:00:00.123456+00:00"] * 3,
+        "elapsed_seconds": 2.530865,
+    }
+    assert events == original
+
+
+@pytest.mark.parametrize("timer_time,elapsed", [
+    ("2026-11-01T00:00:03.123456Z", 0),
+    ("2026-11-01T00:00:02.123455Z", -1.000001),
+    ("2026-11-01T02:00:05.654321+02:00", 2.530865),
+])
+def test_equal_backward_and_offset_times_retain_the_exact_consumed_instant(timer_time: str, elapsed: float) -> None:
+    events = history()
+    events[-1]["recorded_at"] = timer_time
+    value = result(events)
+    assert value["elapsed_seconds"] == elapsed
+    assert datetime.fromisoformat(value["clock"][2]).utcoffset() == timedelta(0)
+
+
+@pytest.mark.parametrize("bad", [None, False, 17, "", "not-a-time", "2026-11-01T00:00:03",
+                                       "2026-02-30T00:00:00Z", "2026-11-01T24:00:00Z",
+                                       "2026-11-01T00:00:03+01:99"])
+def test_invalid_required_time_is_unavailable_then_a_new_proved_boundary_can_recover(bad: Any) -> None:
+    events = history()
+    events[2]["recorded_at"] = bad
+    assert result(events)["clock"] == [None, None, "2026-11-01T00:00:05.654321+00:00"]
+
+
+def test_invalid_present_timestamp_does_not_fall_back_to_recorded_at() -> None:
+    events = history()
+    events[2]["timestamp"] = "invalid"
+    assert result(events)["clock"][1] is None
+
+
+def test_identical_duplicate_terminal_retains_legacy_result_but_has_ambiguous_time() -> None:
+    events = history()
+    events.insert(3, deepcopy(events[2]))
+    assert result(events)["clock"][1] is None
+
+
+@pytest.mark.parametrize("field,value", [("activity_execution_id", "other"), ("sequence", True)])
+def test_unproved_resource_or_counter_keeps_time_unavailable(field: str, value: Any) -> None:
+    events = history()
+    events[2]["payload"][field] = value
+    assert result(events)["clock"][1] is None
+
+
+@pytest.mark.parametrize("terminal", ["ActivityFailed", "ActivityTimedOut", "ActivityCancelled"])
+def test_consumed_failure_uses_its_own_time_and_preserves_failure_delivery(terminal: str) -> None:
+    class Failed:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            try:
+                yield ctx.schedule_activity("clock.first", [])
+            except ActivityFailed:
+                return clock_text(ctx)
+            raise AssertionError("failure was not thrown")
+
+    events = history()[:3]
+    events[-1]["event_type"] = terminal
+    events[-1]["payload"].update(message="failed", reused_recorded_at="2020-01-01T00:00:00Z")
+    assert result(events, Failed) == "2026-11-01T00:00:03.123456+00:00"
+
+
+def test_existing_command_mismatch_remains_the_original_replay_error() -> None:
+    events = history()
+    events[1]["payload"]["activity_type"] = "changed"
+    events[2]["payload"]["activity_type"] = "changed"
+    with pytest.raises(NonDeterministicReplayError):
+        result(events)
+
+
+def test_accessor_is_unavailable_outside_execution_even_with_an_explicit_legacy_clock() -> None:
+    ctx = WorkflowContext(current_time=datetime.fromisoformat("2026-11-01T00:00:00+00:00"))
+    with pytest.raises(ReplayClockUnavailable):
+        ctx.replay_now()
+
+
+def test_supported_query_prefix_is_immutable_and_unknown_final_prefix_is_unavailable() -> None:
+    events = history()
+    original = deepcopy(events)
+    for _ in range(2):
+        assert query_state(ScalarBusinessClockWorkflow, events, [], "clock", payload_codec="avro") == (
+            "2026-11-01T00:00:05.654321+00:00"
+        )
+    assert events == original
+    events.append({"event_type": "WorkflowTaskCompleted", "recorded_at": "2026-12-01T00:00:00Z", "payload": {}})
+    assert query_state(ScalarBusinessClockWorkflow, events, [], "clock", payload_codec="avro") is None
+
+
+@pytest.mark.asyncio
+async def test_async_queries_restore_scope_and_cannot_lend_time_to_a_child_task_or_thread() -> None:
+    first, second = history(), history()
+    second[-1]["recorded_at"] = "2026-11-02T00:00:00Z"
+    values = await asyncio.gather(*(
+            _prepare_query_state(ScalarBusinessClockWorkflow, rows, [], "async_clock", payload_codec="avro")()
+        for rows in (first, second)
+    ))
+    assert values == [
+        ("2026-11-01T00:00:05.654321+00:00", None, None, "2026-11-01T00:00:05.654321+00:00"),
+        ("2026-11-02T00:00:00.000000+00:00", None, None, "2026-11-02T00:00:00.000000+00:00"),
+    ]
+
+
+def test_deadline_loop_advances_after_a_consumed_timer_and_repeated_reads_are_equal() -> None:
+    class Deadline:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("clock.first", [])
+            before = ctx.replay_now()
+            deadline = before + timedelta(seconds=2)
+            steps = 0
+            while ctx.replay_now() < deadline:
+                assert ctx.replay_now() == ctx.replay_now()
+                yield ctx.start_timer(2)
+                steps += 1
+            return steps, (ctx.replay_now() - before).total_seconds()
+
+    assert result(history(), Deadline) == (1, 2.530865)
+
+
+def test_nested_reconstruction_and_callback_error_cannot_borrow_time_and_restore_the_parent_scope() -> None:
+    class Outer:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("clock.first", [])
+            before = ctx.replay_now()
+
+            @workflow.defn(name="tests.business.inner")
+            class Inner:
+                def run(self, nested: WorkflowContext) -> str:
+                    with pytest.raises(ReplayClockUnavailable):
+                        ctx.replay_now()
+                    return "inner"
+
+                @workflow.query("fail")
+                def fail(self) -> None:
+                    with pytest.raises(ReplayClockUnavailable):
+                        ctx.replay_now()
+                    raise RuntimeError("query failed")
+
+            with pytest.raises(QueryFailed, match="query failed"):
+                query_state(Inner, history()[:1], [], "fail", payload_codec="avro")
+            assert ctx.replay_now() == before
+            return clock_text(ctx)
+
+    assert result(history()[:3], Outer) == "2026-11-01T00:00:03.123456+00:00"
+
+
+def _redrive_observations(missing: str | None = None) -> list[str | None]:
+    from tests.test_replay_clock import check_redrive_successor, redrive_fixture
+
+    fixture = redrive_fixture("redrive-successor-reused-timer")
+    check_redrive_successor(fixture["history"], fixture["workflow"])
+    observations: list[str | None] = []
+
+    class Redrive:
+        def run(self, ctx: WorkflowContext, name: str, seconds: int):  # type: ignore[no-untyped-def]
+            observations.append(clock_text(ctx))
+            greeting = yield ctx.schedule_activity("tests.redrive.first", [name])
+            observations.append(clock_text(ctx))
+            yield ctx.start_timer(seconds)
+            observations.append(clock_text(ctx))
+            return (yield ctx.schedule_activity("tests.redrive.second", [greeting]))
+
+    if missing is not None:
+        fixture["history"][0]["payload"].pop(missing)
+    task = fixture["workflow"]
+    outcome = replay(Redrive, fixture["history"], task["input"], workflow_id=task["workflow_id"],
+                     run_id=task["run_id"], payload_codec="avro")
+    try:
+        assert len(outcome.commands) == 1
+        assert outcome.commands[0].activity_type == "tests.redrive.second"
+    finally:
+        outcome.close()
+    return observations
+
+
+def test_canonical_redrive_start_and_reused_results_keep_their_original_times() -> None:
+    expected = ["2026-11-03T00:00:00.000000+00:00"] * 2 + ["2026-11-03T00:00:06.000000+00:00"]
+    assert _redrive_observations() == expected
+    assert _redrive_observations() == expected
+
+
+@pytest.mark.parametrize("missing", ["workflow_command_id", "continued_from_run_id", "workflow_link_id",
+                                      "workflow_run_id", "resume_step_sequence"])
+def test_partial_redrive_authority_is_unavailable_without_changing_legacy_commands(missing: str) -> None:
+    assert _redrive_observations(missing) == [None] * 3
+
+
+def test_unpersisted_side_effect_is_unavailable_on_first_authoring_and_cold_replay() -> None:
+    class Inline:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("clock.first", [])
+            before = clock_text(ctx)
+            yield ctx.side_effect(lambda: "recorded")
+            return before, clock_text(ctx)
+
+    events = history()[:3]
+    first = replay(Inline, events, [], payload_codec="avro")
+    try:
+        assert isinstance(first.commands[-1], CompleteWorkflow)
+        assert first.commands[-1].result == ("2026-11-01T00:00:03.123456+00:00", None)
+    finally:
+        first.close()
+    events.append({"event_type": "SideEffectRecorded", "recorded_at": "2026-11-01T00:00:04Z",
+                   "payload": {"sequence": 2, "result": serializer.envelope("recorded")}})
+    assert result(events, Inline) == ("2026-11-01T00:00:03.123456+00:00", None)
+
+
+def test_a_retained_context_is_unavailable_after_success_and_discard_cleanup() -> None:
+    retained: list[WorkflowContext] = []
+
+    class Retain:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            retained.append(ctx)
+            yield ctx.schedule_activity("clock.first", [])
+            return clock_text(ctx)
+
+    assert result(history()[:3], Retain) == "2026-11-01T00:00:03.123456+00:00"
+    with pytest.raises(ReplayClockUnavailable):
+        retained[0].replay_now()
diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py
index 1fb07e15..b5ee9558 100644
--- a/tests/test_replay_regression_corpus.py
+++ b/tests/test_replay_regression_corpus.py
@@ -25,6 +25,7 @@
     commands_to_server_commands,
     query_state,
 )
+from tests.business_clock_workflows import ScalarBusinessClockWorkflow
 from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate
 from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow
 from tests.test_condition_occurrence_id import OccurrenceApproval, SelectionOccurrences
@@ -371,6 +372,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
 
 
 WORKFLOWS = [
+    ScalarBusinessClockWorkflow,
     PendingFinally,
     BlankFailureWorkflow,
     IdentityAcrossReplay,

From 9a6ab39c2b7a1eb6c037a2b509917af8028be1d9 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:04:56 +0000
Subject: [PATCH 2/7] fix: restore replay clock scope without deleting its
 binding

---
 src/durable_workflow/replay_clock.py | 46 +++++++++++++++-------------
 tests/test_business_replay_clock.py  | 36 +++++++++++++++++++++-
 2 files changed, 60 insertions(+), 22 deletions(-)

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index 0cb50fa3..79f683c6 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -11,7 +11,7 @@
 from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any, ParamSpec, TypeVar
+from typing import Any, ParamSpec, TypeGuard, TypeVar
 
 from .errors import ReplayClockUnavailable
 
@@ -44,11 +44,14 @@ class _Frame:
 
 @contextlib.contextmanager
 def execution_scope(owner: object | None, time: datetime | None) -> Iterator[None]:
-    token = _ACTIVE.set(_Frame(owner, time, threading.get_ident(), _task()))
+    previous = _ACTIVE.get()
+    _ACTIVE.set(_Frame(owner, time, threading.get_ident(), _task()))
     try:
         yield
     finally:
-        _ACTIVE.reset(token)
+        # Restore the value without deleting an absent binding during nested
+        # generator finalization. Cleanup must preserve the original exception.
+        _ACTIVE.set(previous)
 
 
 def isolated(function: Callable[_P, _T]) -> Callable[_P, _T]:
@@ -95,7 +98,7 @@ def _sequence(payload: Mapping[str, Any]) -> int | None:
     return value
 
 
-def _identity(value: Any) -> bool:
+def _identity(value: Any) -> TypeGuard[str]:
     return isinstance(value, str) and bool(value.strip())
 
 
@@ -203,23 +206,9 @@ def _scalar_boundary(self, index: int, family: str, sequence: int) -> bool:
         allowed = {opening, *kinds}
         if family == "activity":
             allowed.add("ActivityStarted")
-        previous = -1
-        for position in indices:
-            row = self.events[position]
-            body = row.get("payload")
-            if (not isinstance(body, Mapping) or _kind(row) not in allowed
-                or position > index or position <= previous or body.get(resource) != identity
-                or body.get("execution_mode") not in (None, "remote") or body.get("local_activity") is True
-                or body.get("parallel_group") is not None or body.get("parallel_group_path") not in (None, [])
-                or body.get("selection_group") is not None
-                or body.get("cancellation_scope_id") not in (None, "root")):
-                return False
-            if family == "timer" and (
-                body.get("timer_kind") not in (None, "sleep", "durable_timer")
-                or "condition_wait_id" in body or "signal_wait_id" in body
-            ):
-                return False
-            previous = position
+        if any(position > index or not self._scalar_row(position, family, resource, identity, allowed)
+               for position in indices):
+            return False
         openings = [position for position in indices if _kind(self.events[position]) == opening]
         starts = [position for position in indices if _kind(self.events[position]) == "ActivityStarted"]
         if len(starts) > 1 or (starts and (not openings or starts[0] < openings[0])):
@@ -228,3 +217,18 @@ def _scalar_boundary(self, index: int, family: str, sequence: int) -> bool:
         if len(openings) != 1 and not (family == "activity" and reused and indices == [index]):
             return False
         return not openings or openings[0] < index
+
+    def _scalar_row(self, index: int, family: str, resource: str, identity: str, allowed: set[str]) -> bool:
+        row = self.events[index]
+        body = row.get("payload")
+        if not isinstance(body, Mapping) or _kind(row) not in allowed or body.get(resource) != identity:
+            return False
+        if (body.get("execution_mode") not in (None, "remote") or body.get("local_activity") is True
+            or body.get("parallel_group") is not None or body.get("parallel_group_path") not in (None, [])
+            or body.get("selection_group") is not None
+            or body.get("cancellation_scope_id") not in (None, "root")):
+            return False
+        return family != "timer" or (
+            body.get("timer_kind") in (None, "sleep", "durable_timer")
+            and "condition_wait_id" not in body and "signal_wait_id" not in body
+        )
diff --git a/tests/test_business_replay_clock.py b/tests/test_business_replay_clock.py
index b6488cea..751221ba 100644
--- a/tests/test_business_replay_clock.py
+++ b/tests/test_business_replay_clock.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import gc
 from copy import deepcopy
 from datetime import datetime, timedelta
 from typing import Any
@@ -189,8 +190,9 @@ def fail(self) -> None:
                         ctx.replay_now()
                     raise RuntimeError("query failed")
 
+            nested_history = history()[:1]
             with pytest.raises(QueryFailed, match="query failed"):
-                query_state(Inner, history()[:1], [], "fail", payload_codec="avro")
+                query_state(Inner, nested_history, [], "fail", payload_codec="avro")
             assert ctx.replay_now() == before
             return clock_text(ctx)
 
@@ -270,3 +272,35 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
     assert result(history()[:3], Retain) == "2026-11-01T00:00:03.123456+00:00"
     with pytest.raises(ReplayClockUnavailable):
         retained[0].replay_now()
+
+
+def test_cyclic_discard_cleanup_restores_parent_time_and_preserves_original_replay_errors() -> None:
+    cleanup: list[str | None] = []
+
+    class Discard:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            try:
+                yield ctx.schedule_activity("discard", [])
+            finally:
+                cleanup.append(clock_text(ctx))
+
+    class Parent:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("clock.first", [])
+            before = ctx.replay_now()
+            for _ in range(32):
+                outcome = replay(Discard, [], [], payload_codec="avro")
+                cycle: list[Any] = [outcome]
+                cycle.append(cycle)
+            del outcome, cycle
+            gc.collect()
+            assert cleanup == [None] * 32
+            assert ctx.replay_now() == before
+            mismatch = history()
+            mismatch[2]["payload"]["activity_type"] = "changed"
+            with pytest.raises(NonDeterministicReplayError):
+                replay(ScalarBusinessClockWorkflow, mismatch, [], payload_codec="avro")
+            assert ctx.replay_now() == before
+            return clock_text(ctx)
+
+    assert result(history()[:3], Parent) == "2026-11-01T00:00:03.123456+00:00"

From 089d7a9022f7106208604483f929cc5a9ecefc2d Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:07:41 +0000
Subject: [PATCH 3/7] refactor: separate scalar clock lifecycle matching

---
 src/durable_workflow/replay_clock.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index 79f683c6..f64b7252 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -209,6 +209,11 @@ def _scalar_boundary(self, index: int, family: str, sequence: int) -> bool:
         if any(position > index or not self._scalar_row(position, family, resource, identity, allowed)
                for position in indices):
             return False
+        return self._scalar_lifecycle(indices, index, family, opening, payload)
+
+    def _scalar_lifecycle(
+        self, indices: Sequence[int], index: int, family: str, opening: str, payload: Mapping[str, Any],
+    ) -> bool:
         openings = [position for position in indices if _kind(self.events[position]) == opening]
         starts = [position for position in indices if _kind(self.events[position]) == "ActivityStarted"]
         if len(starts) > 1 or (starts and (not openings or starts[0] < openings[0])):

From 3cbec0f71e24d80d32e94099f6eb45502e39265a Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:27:36 +0000
Subject: [PATCH 4/7] fix: bind native redrive clock to its accepted start
 prefix

---
 src/durable_workflow/replay_clock.py          | 36 ++++++--
 tests/business_clock_workflows.py             |  8 ++
 .../business-clock-native-redrive-entry.json  | 83 +++++++++++++++++++
 tests/native_redrive_fixture.py               | 75 +++++++++++++++++
 tests/redrive_fixture_registry.py             | 11 +++
 tests/test_business_replay_clock.py           | 83 ++++++++++++++++++-
 tests/test_redrive_successor_evidence.py      | 43 +++++++++-
 tests/test_replay_regression_corpus.py        |  3 +-
 8 files changed, 332 insertions(+), 10 deletions(-)
 create mode 100644 tests/fixtures/replay_regressions/business-clock-native-redrive-entry.json
 create mode 100644 tests/native_redrive_fixture.py

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index f64b7252..0831a0a1 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -87,6 +87,10 @@ def _kind(event: Mapping[str, Any]) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _recorded_value(event: Mapping[str, Any]) -> Any:
+    return event.get("timestamp") if "timestamp" in event else event.get("recorded_at")
+
+
 def _sequence(payload: Mapping[str, Any]) -> int | None:
     value = payload.get("sequence")
     if type(value) is not int or value < 1:
@@ -111,14 +115,16 @@ 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._native_prefix = False
         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:
             index = starts[0]
             payload = events[index].get("payload")
-            if (index == 0 and isinstance(payload, Mapping)
-                and self._redrive_start(payload, workflow_id, run_id)):
+            if (isinstance(payload, Mapping) and self._start_prefix(index, payload)
+                and self._redrive_start(index, payload, workflow_id, run_id)):
                 self.initial = (index, _time(payload.get("replayed_started_at")))
+                self._native_prefix = index == 1
         self._pending = self.initial
         self._current = self.initial
         for index, event in enumerate(events):
@@ -126,7 +132,18 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id
             if isinstance(payload, Mapping) and (sequence := _sequence(payload)) is not None:
                 self._by_sequence.setdefault(sequence, []).append(index)
 
-    def _redrive_start(self, payload: Mapping[str, Any], workflow_id: str, run_id: str) -> bool:
+    def _start_prefix(self, index: int, payload: Mapping[str, Any]) -> bool:
+        if index == 0:
+            return True
+        if index != 1 or _kind(self.events[0]) != "StartAccepted":
+            return False
+        accepted = self.events[0].get("payload")
+        if not isinstance(accepted, Mapping) or accepted.get("outcome") != "redriven":
+            return False
+        return all(_identity(payload.get(key)) and accepted.get(key) == payload[key]
+                   for key in ("workflow_instance_id", "workflow_run_id", "workflow_command_id"))
+
+    def _redrive_start(self, index: int, payload: Mapping[str, Any], workflow_id: str, run_id: str) -> bool:
         if payload.get("recovery_kind") != "redrive":
             # Ordinary entry needs run.started_at, not history insertion time.
             return False
@@ -134,6 +151,8 @@ def _redrive_start(self, payload: Mapping[str, Any], workflow_id: str, run_id: s
                 "workflow_link_id")
         if not all(_identity(payload.get(key)) for key in keys):
             return False
+        if payload["workflow_run_id"] == payload["continued_from_run_id"]:
+            return False
         if ((workflow_id and payload["workflow_instance_id"] != workflow_id)
             or (run_id and payload["workflow_run_id"] != run_id)):
             return False
@@ -143,7 +162,7 @@ def _redrive_start(self, payload: Mapping[str, Any], workflow_id: str, run_id: s
         if any(type(event.get("sequence")) is not int or event["sequence"] != index + 1
                for index, event in enumerate(self.events)):
             return False
-        inserted = _time(self.events[0].get("recorded_at"))
+        inserted = _time(_recorded_value(self.events[index]))
         original = _time(payload.get("replayed_started_at"))
         if inserted is None or original is None or original > inserted:
             return False
@@ -173,12 +192,17 @@ def observe(self, index: int | None, family: str, sequence: int) -> None:
             return
         event = self.events[index]
         payload = event["payload"]
-        value = event.get("timestamp") if "timestamp" in event else event.get("recorded_at")
+        value = _recorded_value(event)
         if _kind(event) in {"ActivityCompleted", "TimerFired"} and "reused_recorded_at" in payload:
+            if self._native_prefix and family == "timer":
+                return
             if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run
                 or self._resume_step is None or sequence >= self._resume_step):
                 return
-            if family == "activity" and not _identity(payload.get("reused_activity_execution_id")):
+            if family == "activity" and (
+                not _identity(payload.get("reused_activity_execution_id"))
+                or payload.get("reused_activity_execution_id") == payload.get("activity_execution_id")
+            ):
                 return
             value = payload["reused_recorded_at"]
             original = _time(value)
diff --git a/tests/business_clock_workflows.py b/tests/business_clock_workflows.py
index 1e29eb73..a6dc3c05 100644
--- a/tests/business_clock_workflows.py
+++ b/tests/business_clock_workflows.py
@@ -58,3 +58,11 @@ async def child_read() -> str | None:
         thread = await asyncio.to_thread(clock_text, self.ctx)
         await asyncio.sleep(0)
         return before, child, thread, clock_text(self.ctx)
+
+
+@workflow.defn(name="tests.replay.business-clock-native-redrive")
+class NativeRedriveBusinessClockWorkflow:
+    def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+        before = clock_text(ctx)
+        yield ctx.schedule_activity("clock.first", [])
+        return {"clock": [before, clock_text(ctx)], "legacy": ctx.now().isoformat(timespec="microseconds")}
diff --git a/tests/fixtures/replay_regressions/business-clock-native-redrive-entry.json b/tests/fixtures/replay_regressions/business-clock-native-redrive-entry.json
new file mode 100644
index 00000000..c0c3fd18
--- /dev/null
+++ b/tests/fixtures/replay_regressions/business-clock-native-redrive-entry.json
@@ -0,0 +1,83 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "business-clock-native-redrive-entry",
+  "protocol_version": "1.20",
+  "bindings": [
+    "python"
+  ],
+  "workflow": {
+    "type": "tests.replay.business-clock-native-redrive",
+    "input": [],
+    "payload_codec": "avro",
+    "workflow_id": "regression-corpus-redrive-successor-timer",
+    "run_id": "01JBREDRIVESUCCESSORRUN0000"
+  },
+  "history": [
+    {
+      "event_type": "StartAccepted",
+      "recorded_at": "2026-11-03T00:04:59.999999+00:00",
+      "payload": {
+        "workflow_instance_id": "regression-corpus-redrive-successor-timer",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "workflow_command_id": "01JBREDRIVECOMMAND00000000",
+        "workflow_type": "tests.replay.business-clock-native-redrive",
+        "workflow_class": "tests.replay.business-clock-native-redrive",
+        "outcome": "redriven"
+      },
+      "sequence": 1
+    },
+    {
+      "sequence": 2,
+      "event_type": "WorkflowStarted",
+      "recorded_at": "2026-11-03T00:05:00+00:00",
+      "payload": {
+        "workflow_type": "tests.replay.business-clock-native-redrive",
+        "workflow_instance_id": "regression-corpus-redrive-successor-timer",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "workflow_command_id": "01JBREDRIVECOMMAND00000000",
+        "continued_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "resume_step_sequence": 2,
+        "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",
+        "workflow_class": "tests.replay.business-clock-native-redrive"
+      }
+    },
+    {
+      "sequence": 3,
+      "event_type": "ActivityCompleted",
+      "recorded_at": "2026-11-03T00:05:00.000001+00:00",
+      "payload": {
+        "activity_execution_id": "01JBREDRIVESUCCESSORACT000",
+        "activity_type": "clock.first",
+        "sequence": 1,
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoKZmlyc3Q="
+        },
+        "payload_codec": "avro",
+        "reused_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "reused_activity_execution_id": "01JBREDRIVESOURCEACTIVITY0",
+        "reused_recorded_at": "2026-11-03T00:00:01.123456+00:00",
+        "activity_class": "clock.first"
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "result": {
+          "clock": [
+            "2026-11-03T00:00:00.000000+00:00",
+            "2026-11-03T00:00:01.123456+00:00"
+          ],
+          "legacy": "2026-11-03T00:00:00.000000+00:00"
+        }
+      }
+    ]
+  }
+}
diff --git a/tests/native_redrive_fixture.py b/tests/native_redrive_fixture.py
new file mode 100644
index 00000000..16b05350
--- /dev/null
+++ b/tests/native_redrive_fixture.py
@@ -0,0 +1,75 @@
+"""Closed activity-prefix evidence for Server 3d4d4afa's native redrive writer.
+
+The producer is rust/src/runtime/redrive.rs, blob
+004836fcd1b9e39255d3cad6a4b0eec01d14aff7. This profile validates the
+retained StartAccepted row explicitly. It does not qualify Core timer reuse.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Any
+
+
+def _require(value: Any, reason: str) -> None:
+    if not value:
+        raise AssertionError(reason)
+
+
+def _time(value: Any) -> datetime:
+    _require(isinstance(value, str), "recorded instant")
+    _require(re.fullmatch(
+        r"[0-9]{4}-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]"
+        r"(?:\.[0-9]{1,6})?(?:Z|\+00:00)", value,
+    ), "native UTC timestamp shape")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    _require(parsed.tzinfo is not None and parsed.utcoffset() is not None, "aware instant")
+    return parsed
+
+
+def check_native_redrive_successor(history: list[dict[str, Any]], workflow: Mapping[str, Any]) -> None:
+    _require(len(history) >= 2, "native redrive prefix")
+    for index, event in enumerate(history, start=1):
+        _require(type(event.get("sequence")) is int and event["sequence"] == index, "native history order")
+        _time(event.get("recorded_at"))
+    accepted, started = history[:2]
+    _require(accepted.get("event_type") == "StartAccepted", "native acceptance prefix")
+    _require(started.get("event_type") == "WorkflowStarted", "native start boundary")
+    start, acceptance = started["payload"], accepted["payload"]
+    _require(isinstance(start, Mapping) and isinstance(acceptance, Mapping), "native start payloads")
+    _require(start.get("recovery_kind") == "redrive", "native recovery kind")
+    _require(acceptance.get("outcome") == "redriven", "native accepted outcome")
+    for key in ("workflow_instance_id", "workflow_run_id", "workflow_command_id", "continued_from_run_id",
+                "workflow_link_id"):
+        _require(isinstance(start.get(key), str) and bool(start[key].strip()), key)
+    for key in ("workflow_instance_id", "workflow_run_id", "workflow_command_id"):
+        _require(acceptance.get(key) == start[key], "native accepted identity")
+    _require(start["workflow_instance_id"] == workflow.get("workflow_id"), "native fixture instance")
+    _require(start["workflow_run_id"] == workflow.get("run_id"), "native fixture run")
+    _require(start.get("workflow_type") == workflow.get("type"), "native fixture type")
+    _require(acceptance.get("workflow_type") == workflow.get("type"), "native accepted type")
+    _require(acceptance.get("workflow_class") == workflow.get("type"), "native accepted class")
+    _require(start["workflow_run_id"] != start["continued_from_run_id"], "native successor is a new run")
+    inserted = _time(started["recorded_at"])
+    _require(_time(start.get("replayed_started_at")) <= inserted, "native original start")
+    _require(type(start.get("resume_step_sequence")) is int
+             and start["resume_step_sequence"] == len(history) - 1, "native resume after prefix")
+    identities: set[str] = set()
+    for sequence, event in enumerate(history[2:], start=1):
+        _require(event.get("event_type") == "ActivityCompleted", "native activity-only reused prefix")
+        payload = event["payload"]
+        _require(type(payload.get("sequence")) is int and payload["sequence"] == sequence,
+                 "native logical prefix order")
+        for key in ("activity_execution_id", "reused_activity_execution_id", "activity_type"):
+            _require(isinstance(payload.get(key), str) and bool(payload[key].strip()), key)
+        identity = payload["activity_execution_id"]
+        _require(identity not in identities and identity != payload["reused_activity_execution_id"],
+                 "native fresh activity identity")
+        identities.add(identity)
+        _require(payload.get("reused_from_run_id") == start["continued_from_run_id"], "native source run")
+        _require(payload.get("payload_codec") == "avro" and "result" in payload, "native Avro result")
+        result = payload["result"]
+        _require(not isinstance(result, Mapping) or "external_payload" not in result, "native inline result")
+        _require(_time(payload.get("reused_recorded_at")) <= inserted, "native original result time")
diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py
index fda7570e..0131bd89 100644
--- a/tests/redrive_fixture_registry.py
+++ b/tests/redrive_fixture_registry.py
@@ -7,6 +7,10 @@
 named error, and its named replacement must pass. The gate in
 ``tests/test_redrive_successor_evidence.py`` requires every fixture that
 looks like a redrive to be registered here.
+
+Native successors have a separate closed activity-only writer profile. Every
+such fixture must pass that profile, while the existing Core checker and all
+of its positive/negative expectations remain unchanged.
 """
 
 from __future__ import annotations
@@ -40,6 +44,13 @@ class NonQualifying:
     "redrive-successor-reused-upserts-cancellation-clock": 3,
 }
 
+# Native redrive retains a same-identity StartAccepted row and reuses only
+# activities. Its separate closed writer profile must pass; Core's checker
+# remains unchanged and must continue to reject this different shape.
+NATIVE_SUCCESSORS: dict[str, int] = {
+    "business-clock-native-redrive-entry": 2,
+}
+
 NON_QUALIFYING: dict[str, NonQualifying] = {
     "redrive-start-replayed-clock": NonQualifying(
         sha256="70653171faf5ddba5ed264d0a7ce8bb1af97112dbd6e3bb0e39c6f155fee0fd2",
diff --git a/tests/test_business_replay_clock.py b/tests/test_business_replay_clock.py
index 751221ba..5692824b 100644
--- a/tests/test_business_replay_clock.py
+++ b/tests/test_business_replay_clock.py
@@ -11,7 +11,7 @@
 from durable_workflow import ReplayClockUnavailable, serializer, workflow
 from durable_workflow.errors import ActivityFailed, NonDeterministicReplayError, QueryFailed
 from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, _prepare_query_state, query_state, replay
-from tests.business_clock_workflows import ScalarBusinessClockWorkflow, clock_text
+from tests.business_clock_workflows import NativeRedriveBusinessClockWorkflow, ScalarBusinessClockWorkflow, clock_text
 
 
 def history() -> list[dict[str, Any]]:
@@ -304,3 +304,84 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
             return clock_text(ctx)
 
     assert result(history()[:3], Parent) == "2026-11-01T00:00:03.123456+00:00"
+
+
+def _native_fixture() -> dict[str, Any]:
+    from tests.native_redrive_fixture import check_native_redrive_successor
+    from tests.test_replay_clock import redrive_fixture
+
+    fixture = redrive_fixture("business-clock-native-redrive-entry")
+    check_native_redrive_successor(fixture["history"], fixture["workflow"])
+    return fixture
+
+
+@pytest.mark.parametrize("public", [False, True])
+def test_native_same_identity_acceptance_prefix_and_both_time_representations_are_supported(public: bool) -> None:
+    fixture = _native_fixture()
+    events = fixture["history"]
+    if public:
+        for event in events:
+            event["timestamp"] = event.pop("recorded_at")
+    assert result(events, NativeRedriveBusinessClockWorkflow) == fixture["expected"]["command_sequence"][0]["result"]
+
+
+@pytest.mark.parametrize("mutation", ["prefix", "instance", "run", "command", "outcome", "extra_prefix",
+                                      "duplicate_start", "order", "self_source", "start_time"])
+def test_native_entry_refuses_unrelated_prefixes_conflicting_identity_and_ambiguous_start(mutation: str) -> None:
+    events = _native_fixture()["history"]
+    if mutation == "prefix":
+        events[0]["event_type"] = "WorkflowTaskCompleted"
+    elif mutation in {"instance", "run", "command"}:
+        key = {"instance": "workflow_instance_id", "run": "workflow_run_id", "command": "workflow_command_id"}[mutation]
+        events[0]["payload"][key] = "foreign"
+    elif mutation == "outcome":
+        events[0]["payload"]["outcome"] = "started"
+    elif mutation == "extra_prefix":
+        events.insert(0, {"event_type": "WorkflowTaskCompleted", "payload": {}})
+    elif mutation == "duplicate_start":
+        events.insert(2, deepcopy(events[1]))
+    elif mutation == "self_source":
+        events[1]["payload"]["continued_from_run_id"] = events[1]["payload"]["workflow_run_id"]
+        events[-1]["payload"]["reused_from_run_id"] = events[1]["payload"]["workflow_run_id"]
+    elif mutation == "start_time":
+        events[1]["timestamp"] = None
+    for index, event in enumerate(events, start=1):
+        event["sequence"] = index
+    if mutation == "order":
+        events[-1]["sequence"] = 9
+    assert result(events, NativeRedriveBusinessClockWorkflow)["clock"] == [None, None]
+
+
+def test_an_unproved_reused_identity_does_not_poison_the_earlier_proved_start() -> None:
+    fixture = _native_fixture()
+    events = fixture["history"]
+    events[-1]["payload"]["activity_execution_id"] = events[-1]["payload"]["reused_activity_execution_id"]
+    first = fixture["expected"]["command_sequence"][0]["result"]["clock"][0]
+    assert result(events, NativeRedriveBusinessClockWorkflow)["clock"] == [first, None]
+
+
+def test_native_acceptance_prefix_does_not_qualify_core_only_reused_timers() -> None:
+    from tests.test_replay_clock import redrive_fixture
+
+    fixture = redrive_fixture("redrive-successor-reused-timer")
+    events = fixture["history"]
+    accepted = _native_fixture()["history"][0]
+    accepted["payload"].update(workflow_type=events[0]["payload"]["workflow_type"],
+                               workflow_class=events[0]["payload"]["workflow_type"])
+    events.insert(0, accepted)
+    for index, event in enumerate(events, start=1):
+        event["sequence"] = index
+
+    class MixedProducer:
+        def run(self, ctx: WorkflowContext, name: str, seconds: int):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("tests.redrive.first", [name])
+            yield ctx.start_timer(seconds)
+            return clock_text(ctx)
+
+    task = fixture["workflow"]
+    outcome = replay(MixedProducer, events, task["input"], workflow_id=task["workflow_id"],
+                     run_id=task["run_id"], payload_codec="avro")
+    try:
+        assert outcome.commands == [CompleteWorkflow(None)]
+    finally:
+        outcome.close()
diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py
index 4d2617bc..dbd64bcf 100644
--- a/tests/test_redrive_successor_evidence.py
+++ b/tests/test_redrive_successor_evidence.py
@@ -24,7 +24,8 @@
 from durable_workflow import workflow
 from durable_workflow.errors import WorkflowCancelled
 from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, WorkflowContext, query_state, replay
-from tests.redrive_fixture_registry import NON_QUALIFYING, SOURCE_RUNS, VALID_SUCCESSORS
+from tests.native_redrive_fixture import check_native_redrive_successor
+from tests.redrive_fixture_registry import NATIVE_SUCCESSORS, NON_QUALIFYING, SOURCE_RUNS, VALID_SUCCESSORS
 from tests.test_replay_clock import REDRIVE_FIXTURES, check_redrive_successor, redrive_fixture, redrive_time
 
 SIDE_EFFECT_CALLS: list[str] = []
@@ -126,7 +127,45 @@ def _verdict(name: str) -> str | None:
 
 def test_every_redrive_fixture_is_registered() -> None:
     selected = {path.stem for path in REDRIVE_FIXTURES.glob("*.json") if _looks_like_redrive(path)}
-    assert selected == set(VALID_SUCCESSORS) | set(NON_QUALIFYING) | SOURCE_RUNS
+    assert selected == set(VALID_SUCCESSORS) | set(NATIVE_SUCCESSORS) | set(NON_QUALIFYING) | SOURCE_RUNS
+
+
+@pytest.mark.parametrize("name", sorted(NATIVE_SUCCESSORS))
+def test_native_successor_profile_is_closed_and_does_not_inherit_core_evidence(name: str) -> None:
+    fixture = redrive_fixture(name)
+    check_native_redrive_successor(fixture["history"], fixture["workflow"])
+    assert fixture["history"][1]["payload"]["resume_step_sequence"] == NATIVE_SUCCESSORS[name]
+    assert _verdict(name) is not None
+
+
+@pytest.mark.parametrize("mutation", ["prefix", "instance", "run", "command", "outcome", "order",
+                                      "resume", "source", "duplicate", "timer", "external", "time"])
+def test_native_successor_profile_rejects_foreign_and_unsupported_shapes(mutation: str) -> None:
+    fixture = redrive_fixture("business-clock-native-redrive-entry")
+    history = fixture["history"]
+    if mutation == "prefix":
+        history[0]["event_type"] = "WorkflowTaskCompleted"
+    elif mutation in {"instance", "run", "command"}:
+        key = {"instance": "workflow_instance_id", "run": "workflow_run_id", "command": "workflow_command_id"}[mutation]
+        history[0]["payload"][key] = "foreign"
+    elif mutation == "outcome":
+        history[0]["payload"]["outcome"] = "started"
+    elif mutation == "order":
+        history[-1]["sequence"] = 9
+    elif mutation == "resume":
+        history[1]["payload"]["resume_step_sequence"] = True
+    elif mutation == "source":
+        history[-1]["payload"]["reused_from_run_id"] = "foreign"
+    elif mutation == "duplicate":
+        history.append(deepcopy(history[1]))
+    elif mutation == "timer":
+        history[-1]["event_type"] = "TimerFired"
+    elif mutation == "external":
+        history[-1]["payload"]["result"] = {"codec": "avro", "external_payload": {}}
+    else:
+        history[1]["recorded_at"] = "2026-11-03T24:00:00Z"
+    with pytest.raises((AssertionError, ValueError)):
+        check_native_redrive_successor(history, fixture["workflow"])
 
 
 def test_exactly_the_registered_successors_pass() -> None:
diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py
index b5ee9558..b2dd9816 100644
--- a/tests/test_replay_regression_corpus.py
+++ b/tests/test_replay_regression_corpus.py
@@ -25,7 +25,7 @@
     commands_to_server_commands,
     query_state,
 )
-from tests.business_clock_workflows import ScalarBusinessClockWorkflow
+from tests.business_clock_workflows import NativeRedriveBusinessClockWorkflow, ScalarBusinessClockWorkflow
 from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate
 from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow
 from tests.test_condition_occurrence_id import OccurrenceApproval, SelectionOccurrences
@@ -372,6 +372,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
 
 
 WORKFLOWS = [
+    NativeRedriveBusinessClockWorkflow,
     ScalarBusinessClockWorkflow,
     PendingFinally,
     BlankFailureWorkflow,

From 199caf6c57f87c46f0bd1084a082a3965ad53b51 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:31:10 +0000
Subject: [PATCH 5/7] refactor: isolate reused business time validation

---
 src/durable_workflow/replay_clock.py | 36 +++++++++++++++-------------
 1 file changed, 20 insertions(+), 16 deletions(-)

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index 0831a0a1..2688a050 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -192,23 +192,27 @@ def observe(self, index: int | None, family: str, sequence: int) -> None:
             return
         event = self.events[index]
         payload = event["payload"]
-        value = _recorded_value(event)
         if _kind(event) in {"ActivityCompleted", "TimerFired"} and "reused_recorded_at" in payload:
-            if self._native_prefix and family == "timer":
-                return
-            if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run
-                or self._resume_step is None or sequence >= self._resume_step):
-                return
-            if family == "activity" and (
-                not _identity(payload.get("reused_activity_execution_id"))
-                or payload.get("reused_activity_execution_id") == payload.get("activity_execution_id")
-            ):
-                return
-            value = payload["reused_recorded_at"]
-            original = _time(value)
-            if original is None or self._redrive_inserted is None or original > self._redrive_inserted:
-                return
-        self._pending = (index, _time(value))
+            time = self._reused_time(payload, family, sequence)
+        else:
+            time = _time(_recorded_value(event))
+        self._pending = (index, time)
+
+    def _reused_time(self, payload: Mapping[str, Any], family: str, sequence: int) -> datetime | None:
+        if self._native_prefix and family == "timer":
+            return None
+        if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run
+            or self._resume_step is None or sequence >= self._resume_step):
+            return None
+        if family == "activity" and (
+            not _identity(payload.get("reused_activity_execution_id"))
+            or payload.get("reused_activity_execution_id") == payload.get("activity_execution_id")
+        ):
+            return None
+        original = _time(payload.get("reused_recorded_at"))
+        if original is None or self._redrive_inserted is None or original > self._redrive_inserted:
+            return None
+        return original
 
     def _scalar_boundary(self, index: int, family: str, sequence: int) -> bool:
         event = self.events[index]

From 3718fbf690779fd8d1e6b5b6ff3f2281d6a544fd Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:42:44 +0000
Subject: [PATCH 6/7] fix: preserve Core timer time behind the shared redrive
 prefix

---
 src/durable_workflow/replay_clock.py          |   4 -
 tests/business_clock_workflows.py             |  11 ++
 tests/core_redrive_fixture.py                 |  36 ++++++
 ...s-clock-core-redrive-acceptance-timer.json | 107 ++++++++++++++++++
 tests/redrive_fixture_registry.py             |   6 +
 tests/test_business_replay_clock.py           |   8 +-
 tests/test_redrive_successor_evidence.py      |  46 +++++++-
 tests/test_replay_regression_corpus.py        |   7 +-
 8 files changed, 214 insertions(+), 11 deletions(-)
 create mode 100644 tests/core_redrive_fixture.py
 create mode 100644 tests/fixtures/replay_regressions/business-clock-core-redrive-acceptance-timer.json

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index 2688a050..425e250b 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -115,7 +115,6 @@ 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._native_prefix = False
         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:
@@ -124,7 +123,6 @@ def __init__(self, events: Sequence[Mapping[str, Any]], workflow_id: str, run_id
             if (isinstance(payload, Mapping) and self._start_prefix(index, payload)
                 and self._redrive_start(index, payload, workflow_id, run_id)):
                 self.initial = (index, _time(payload.get("replayed_started_at")))
-                self._native_prefix = index == 1
         self._pending = self.initial
         self._current = self.initial
         for index, event in enumerate(events):
@@ -199,8 +197,6 @@ def observe(self, index: int | None, family: str, sequence: int) -> None:
         self._pending = (index, time)
 
     def _reused_time(self, payload: Mapping[str, Any], family: str, sequence: int) -> datetime | None:
-        if self._native_prefix and family == "timer":
-            return None
         if (self._source_run is None or payload.get("reused_from_run_id") != self._source_run
             or self._resume_step is None or sequence >= self._resume_step):
             return None
diff --git a/tests/business_clock_workflows.py b/tests/business_clock_workflows.py
index a6dc3c05..cf620d5b 100644
--- a/tests/business_clock_workflows.py
+++ b/tests/business_clock_workflows.py
@@ -66,3 +66,14 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
         before = clock_text(ctx)
         yield ctx.schedule_activity("clock.first", [])
         return {"clock": [before, clock_text(ctx)], "legacy": ctx.now().isoformat(timespec="microseconds")}
+
+
+@workflow.defn(name="tests.replay.business-clock-core-prefix-timer")
+class PrefixedCoreTimerBusinessClockWorkflow:
+    def run(self, ctx: WorkflowContext, name: str, seconds: int):  # type: ignore[no-untyped-def]
+        observations = [clock_text(ctx)]
+        yield ctx.schedule_activity("tests.redrive.first", [name])
+        observations.append(clock_text(ctx))
+        yield ctx.start_timer(seconds)
+        observations.append(clock_text(ctx))
+        return {"clock": observations, "legacy": ctx.now().isoformat(timespec="microseconds")}
diff --git a/tests/core_redrive_fixture.py b/tests/core_redrive_fixture.py
new file mode 100644
index 00000000..28883be2
--- /dev/null
+++ b/tests/core_redrive_fixture.py
@@ -0,0 +1,36 @@
+"""Core's retained acceptance row plus the unchanged successor checker.
+
+DefaultWorkflowControlPlane at 98bd9868 (blob af6ac905847db3f5b8fd43d0e7f825475a7b6318)
+emits this prefix. Removing only that validated row gives the existing Core
+successor representation; timestamps and payloads are never rewritten.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from copy import deepcopy
+from typing import Any
+
+
+def check_prefixed_core_successor(history: list[dict[str, Any]], workflow: Mapping[str, Any]) -> None:
+    from tests.test_replay_clock import check_redrive_successor, redrive_time
+
+    assert len(history) >= 3, "Core accepted successor"
+    acceptance, started = history[:2]
+    assert acceptance.get("event_type") == "StartAccepted", "Core acceptance prefix"
+    assert started.get("event_type") == "WorkflowStarted", "Core start boundary"
+    accepted, start = acceptance["payload"], started["payload"]
+    assert isinstance(accepted, Mapping) and isinstance(start, Mapping), "Core start payloads"
+    assert accepted.get("outcome") == "redriven", "Core accepted outcome"
+    for key in ("workflow_instance_id", "workflow_run_id", "workflow_command_id"):
+        assert isinstance(start.get(key), str) and start[key].strip(), key
+        assert accepted.get(key) == start[key], "Core accepted identity"
+    assert accepted.get("workflow_type") == workflow.get("type"), "Core accepted workflow type"
+    assert isinstance(accepted.get("workflow_class"), str) and accepted["workflow_class"].strip(), "Core class"
+    redrive_time(acceptance["recorded_at"])
+    for index, event in enumerate(history, start=1):
+        assert type(event.get("sequence")) is int and event["sequence"] == index, "Core complete history order"
+    successor = deepcopy(history[1:])
+    for index, event in enumerate(successor, start=1):
+        event["sequence"] = index
+    check_redrive_successor(successor, workflow)
diff --git a/tests/fixtures/replay_regressions/business-clock-core-redrive-acceptance-timer.json b/tests/fixtures/replay_regressions/business-clock-core-redrive-acceptance-timer.json
new file mode 100644
index 00000000..d45fe468
--- /dev/null
+++ b/tests/fixtures/replay_regressions/business-clock-core-redrive-acceptance-timer.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-core-redrive-acceptance-timer",
+  "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": [
+    {
+      "event_type": "StartAccepted",
+      "recorded_at": "2026-11-03T00:05:00+00:00",
+      "payload": {
+        "workflow_instance_id": "regression-corpus-redrive-successor-timer",
+        "workflow_run_id": "01JBREDRIVESUCCESSORRUN0000",
+        "workflow_command_id": "01JBREDRIVECOMMAND00000000",
+        "workflow_type": "tests.replay.business-clock-core-prefix-timer",
+        "workflow_class": "Tests\\Fixtures\\BusinessClockCoreTimerWorkflow",
+        "outcome": "redriven"
+      },
+      "sequence": 1
+    },
+    {
+      "sequence": 2,
+      "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": 3,
+      "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": 4,
+      "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": 5,
+      "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",
+        "reused_from_run_id": "01JBREDRIVESOURCERUN000000",
+        "reused_recorded_at": "2026-11-03T00:00:06+00:00"
+      }
+    }
+  ],
+  "expected": {
+    "command_sequence": [
+      {
+        "type": "complete_workflow",
+        "result": {
+          "clock": [
+            "2026-11-03T00:00:00.000000+00:00",
+            "2026-11-03T00:00:00.000000+00:00",
+            "2026-11-03T00:00:06.000000+00:00"
+          ],
+          "legacy": "2026-11-03T00:00:00.000000+00:00"
+        }
+      }
+    ]
+  }
+}
diff --git a/tests/redrive_fixture_registry.py b/tests/redrive_fixture_registry.py
index 0131bd89..e349f7b3 100644
--- a/tests/redrive_fixture_registry.py
+++ b/tests/redrive_fixture_registry.py
@@ -51,6 +51,12 @@ class NonQualifying:
     "business-clock-native-redrive-entry": 2,
 }
 
+# The complete Core writer retains the same acceptance row as native. Its
+# finite prefix adapter must validate that row before the unchanged Core gate.
+ACCEPTED_CORE_SUCCESSORS: dict[str, int] = {
+    "business-clock-core-redrive-acceptance-timer": 3,
+}
+
 NON_QUALIFYING: dict[str, NonQualifying] = {
     "redrive-start-replayed-clock": NonQualifying(
         sha256="70653171faf5ddba5ed264d0a7ce8bb1af97112dbd6e3bb0e39c6f155fee0fd2",
diff --git a/tests/test_business_replay_clock.py b/tests/test_business_replay_clock.py
index 5692824b..91f532c3 100644
--- a/tests/test_business_replay_clock.py
+++ b/tests/test_business_replay_clock.py
@@ -360,7 +360,7 @@ def test_an_unproved_reused_identity_does_not_poison_the_earlier_proved_start()
     assert result(events, NativeRedriveBusinessClockWorkflow)["clock"] == [first, None]
 
 
-def test_native_acceptance_prefix_does_not_qualify_core_only_reused_timers() -> None:
+def test_complete_core_acceptance_prefix_retains_reused_timer_time() -> None:
     from tests.test_replay_clock import redrive_fixture
 
     fixture = redrive_fixture("redrive-successor-reused-timer")
@@ -372,16 +372,16 @@ def test_native_acceptance_prefix_does_not_qualify_core_only_reused_timers() ->
     for index, event in enumerate(events, start=1):
         event["sequence"] = index
 
-    class MixedProducer:
+    class CoreTimer:
         def run(self, ctx: WorkflowContext, name: str, seconds: int):  # type: ignore[no-untyped-def]
             yield ctx.schedule_activity("tests.redrive.first", [name])
             yield ctx.start_timer(seconds)
             return clock_text(ctx)
 
     task = fixture["workflow"]
-    outcome = replay(MixedProducer, events, task["input"], workflow_id=task["workflow_id"],
+    outcome = replay(CoreTimer, events, task["input"], workflow_id=task["workflow_id"],
                      run_id=task["run_id"], payload_codec="avro")
     try:
-        assert outcome.commands == [CompleteWorkflow(None)]
+        assert outcome.commands == [CompleteWorkflow("2026-11-03T00:00:06.000000+00:00")]
     finally:
         outcome.close()
diff --git a/tests/test_redrive_successor_evidence.py b/tests/test_redrive_successor_evidence.py
index dbd64bcf..f332ce12 100644
--- a/tests/test_redrive_successor_evidence.py
+++ b/tests/test_redrive_successor_evidence.py
@@ -24,8 +24,15 @@
 from durable_workflow import workflow
 from durable_workflow.errors import WorkflowCancelled
 from durable_workflow.workflow import CompleteWorkflow, ScheduleActivity, WorkflowContext, query_state, replay
+from tests.core_redrive_fixture import check_prefixed_core_successor
 from tests.native_redrive_fixture import check_native_redrive_successor
-from tests.redrive_fixture_registry import NATIVE_SUCCESSORS, NON_QUALIFYING, SOURCE_RUNS, VALID_SUCCESSORS
+from tests.redrive_fixture_registry import (
+    ACCEPTED_CORE_SUCCESSORS,
+    NATIVE_SUCCESSORS,
+    NON_QUALIFYING,
+    SOURCE_RUNS,
+    VALID_SUCCESSORS,
+)
 from tests.test_replay_clock import REDRIVE_FIXTURES, check_redrive_successor, redrive_fixture, redrive_time
 
 SIDE_EFFECT_CALLS: list[str] = []
@@ -127,7 +134,42 @@ def _verdict(name: str) -> str | None:
 
 def test_every_redrive_fixture_is_registered() -> None:
     selected = {path.stem for path in REDRIVE_FIXTURES.glob("*.json") if _looks_like_redrive(path)}
-    assert selected == set(VALID_SUCCESSORS) | set(NATIVE_SUCCESSORS) | set(NON_QUALIFYING) | SOURCE_RUNS
+    assert selected == (
+        set(VALID_SUCCESSORS) | set(NATIVE_SUCCESSORS) | set(ACCEPTED_CORE_SUCCESSORS)
+        | set(NON_QUALIFYING) | SOURCE_RUNS
+    )
+
+
+@pytest.mark.parametrize("name", sorted(ACCEPTED_CORE_SUCCESSORS))
+def test_complete_core_prefix_is_validated_without_changing_the_retained_successor_checker(name: str) -> None:
+    fixture = redrive_fixture(name)
+    original = deepcopy(fixture)
+    check_prefixed_core_successor(fixture["history"], fixture["workflow"])
+    assert fixture["history"][1]["payload"]["resume_step_sequence"] == ACCEPTED_CORE_SUCCESSORS[name]
+    assert fixture == original
+    assert _verdict(name) is not None
+    with pytest.raises(AssertionError):
+        check_native_redrive_successor(fixture["history"], fixture["workflow"])
+
+
+@pytest.mark.parametrize("mutation", ["prefix", "identity", "order", "duplicate", "timer", "outcome"])
+def test_complete_core_prefix_adapter_preserves_identity_order_and_successor_refusals(mutation: str) -> None:
+    fixture = redrive_fixture("business-clock-core-redrive-acceptance-timer")
+    events = fixture["history"]
+    if mutation == "prefix":
+        events[0]["event_type"] = "WorkflowTaskCompleted"
+    elif mutation == "identity":
+        events[0]["payload"]["workflow_command_id"] = "foreign"
+    elif mutation == "order":
+        events[-1]["sequence"] = 99
+    elif mutation == "duplicate":
+        events.insert(2, deepcopy(events[1]))
+    elif mutation == "timer":
+        events[-1]["payload"]["timer_id"] = "foreign"
+    else:
+        events[0]["payload"]["outcome"] = "started"
+    with pytest.raises(AssertionError):
+        check_prefixed_core_successor(events, fixture["workflow"])
 
 
 @pytest.mark.parametrize("name", sorted(NATIVE_SUCCESSORS))
diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py
index b2dd9816..7ed48722 100644
--- a/tests/test_replay_regression_corpus.py
+++ b/tests/test_replay_regression_corpus.py
@@ -25,7 +25,11 @@
     commands_to_server_commands,
     query_state,
 )
-from tests.business_clock_workflows import NativeRedriveBusinessClockWorkflow, ScalarBusinessClockWorkflow
+from tests.business_clock_workflows import (
+    NativeRedriveBusinessClockWorkflow,
+    PrefixedCoreTimerBusinessClockWorkflow,
+    ScalarBusinessClockWorkflow,
+)
 from tests.condition_fingerprint_workflows import FingerprintApproval, FingerprintGenexpr, FingerprintTerminalDuplicate
 from tests.deregistration_recovery import RecoveredWorkflow as FencedRecoveryWorkflow
 from tests.test_condition_occurrence_id import OccurrenceApproval, SelectionOccurrences
@@ -373,6 +377,7 @@ def run(self, ctx: WorkflowContext, message: str) -> None:
 
 WORKFLOWS = [
     NativeRedriveBusinessClockWorkflow,
+    PrefixedCoreTimerBusinessClockWorkflow,
     ScalarBusinessClockWorkflow,
     PendingFinally,
     BlankFailureWorkflow,

From f944108e1d0e72554d10080d7693f5b0afbef265 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 03:57:40 +0000
Subject: [PATCH 7/7] Keep replay disposal out of context variable updates

---
 src/durable_workflow/replay_clock.py | 25 +++++++++---
 src/durable_workflow/workflow.py     |  2 +-
 tests/test_business_replay_clock.py  | 60 +++++++++++++++++++++++++++-
 3 files changed, 79 insertions(+), 8 deletions(-)

diff --git a/src/durable_workflow/replay_clock.py b/src/durable_workflow/replay_clock.py
index 425e250b..017b08a8 100644
--- a/src/durable_workflow/replay_clock.py
+++ b/src/durable_workflow/replay_clock.py
@@ -40,18 +40,16 @@ class _Frame:
 
 
 _ACTIVE: ContextVar[_Frame | None] = ContextVar("business_replay_clock", default=None)
+_DISCARD = threading.local()
 
 
 @contextlib.contextmanager
 def execution_scope(owner: object | None, time: datetime | None) -> Iterator[None]:
-    previous = _ACTIVE.get()
-    _ACTIVE.set(_Frame(owner, time, threading.get_ident(), _task()))
+    token = _ACTIVE.set(_Frame(owner, time, threading.get_ident(), _task()))
     try:
         yield
     finally:
-        # Restore the value without deleting an absent binding during nested
-        # generator finalization. Cleanup must preserve the original exception.
-        _ACTIVE.set(previous)
+        _ACTIVE.reset(token)
 
 
 def isolated(function: Callable[_P, _T]) -> Callable[_P, _T]:
@@ -63,9 +61,24 @@ def invoke(*args: _P.args, **kwargs: _P.kwargs) -> _T:
     return invoke
 
 
+def discarded(function: Callable[_P, _T]) -> Callable[_P, _T]:
+    """Synchronous disposal cannot borrow an active business clock."""
+    @functools.wraps(function)
+    def invoke(*args: _P.args, **kwargs: _P.kwargs) -> _T:
+        # Finalizers can interrupt a ContextVar update. Do not write another
+        # ContextVar here or move generator cleanup away from its own tokens.
+        previous = getattr(_DISCARD, "active", False)
+        _DISCARD.active = True
+        try:
+            return function(*args, **kwargs)
+        finally:
+            _DISCARD.active = previous
+    return invoke
+
+
 def read(owner: object) -> datetime:
     frame = _ACTIVE.get()
-    if (frame is None or frame.owner is not owner or frame.time is None
+    if (getattr(_DISCARD, "active", False) or frame is None or frame.owner is not owner or frame.time is None
         or frame.thread != threading.get_ident() or frame.task is not _task()):
         raise ReplayClockUnavailable()
     return frame.time
diff --git a/src/durable_workflow/workflow.py b/src/durable_workflow/workflow.py
index 118eece0..2dc7a873 100644
--- a/src/durable_workflow/workflow.py
+++ b/src/durable_workflow/workflow.py
@@ -2921,7 +2921,7 @@ def __init__(self, logger: _ReplayLogger) -> None:
         self.generator: Any = None
         self.awaitable: Any = None
 
-    @replay_clock.isolated
+    @replay_clock.discarded
     def close(self) -> None:
         self.logger._set_replaying(True)
         awaitable, self.awaitable = self.awaitable, None
diff --git a/tests/test_business_replay_clock.py b/tests/test_business_replay_clock.py
index 91f532c3..a9088afa 100644
--- a/tests/test_business_replay_clock.py
+++ b/tests/test_business_replay_clock.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import gc
+from contextvars import ContextVar
 from copy import deepcopy
 from datetime import datetime, timedelta
 from typing import Any
@@ -10,7 +11,14 @@
 
 from durable_workflow import ReplayClockUnavailable, serializer, workflow
 from durable_workflow.errors import ActivityFailed, NonDeterministicReplayError, QueryFailed
-from durable_workflow.workflow import CompleteWorkflow, WorkflowContext, _prepare_query_state, query_state, replay
+from durable_workflow.workflow import (
+    _ACTIVE_WORKFLOW_REPLAY,
+    CompleteWorkflow,
+    WorkflowContext,
+    _prepare_query_state,
+    query_state,
+    replay,
+)
 from tests.business_clock_workflows import NativeRedriveBusinessClockWorkflow, ScalarBusinessClockWorkflow, clock_text
 
 
@@ -121,6 +129,7 @@ def test_existing_command_mismatch_remains_the_original_replay_error() -> None:
     events[2]["payload"]["activity_type"] = "changed"
     with pytest.raises(NonDeterministicReplayError):
         result(events)
+    assert _ACTIVE_WORKFLOW_REPLAY.get() is None
 
 
 def test_accessor_is_unavailable_outside_execution_even_with_an_explicit_legacy_clock() -> None:
@@ -288,6 +297,7 @@ class Parent:
         def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
             yield ctx.schedule_activity("clock.first", [])
             before = ctx.replay_now()
+            assert _ACTIVE_WORKFLOW_REPLAY.get() is ctx
             for _ in range(32):
                 outcome = replay(Discard, [], [], payload_codec="avro")
                 cycle: list[Any] = [outcome]
@@ -296,14 +306,62 @@ def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
             gc.collect()
             assert cleanup == [None] * 32
             assert ctx.replay_now() == before
+            assert _ACTIVE_WORKFLOW_REPLAY.get() is ctx
             mismatch = history()
             mismatch[2]["payload"]["activity_type"] = "changed"
             with pytest.raises(NonDeterministicReplayError):
                 replay(ScalarBusinessClockWorkflow, mismatch, [], payload_codec="avro")
             assert ctx.replay_now() == before
+            assert _ACTIVE_WORKFLOW_REPLAY.get() is ctx
             return clock_text(ctx)
 
     assert result(history()[:3], Parent) == "2026-11-01T00:00:03.123456+00:00"
+    assert _ACTIVE_WORKFLOW_REPLAY.get() is None
+
+
+@pytest.mark.parametrize("collected", [False, True], ids=["explicit-close", "cyclic-finalizer"])
+def test_discard_keeps_generator_context_tokens_and_cannot_borrow_parent_time(collected: bool) -> None:
+    ambient: ContextVar[str] = ContextVar("clock_cleanup_ambient", default="outside")
+    observations: list[Any] = []
+    parent: list[WorkflowContext] = []
+
+    class Discard:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            token = ambient.set("suspended")
+            try:
+                yield ctx.schedule_activity("discard", [])
+            finally:
+                observations.append((ambient.get(), clock_text(ctx), clock_text(parent[0])))
+                ambient.reset(token)
+                observations.append(ambient.get())
+
+    class Parent:
+        def run(self, ctx: WorkflowContext):  # type: ignore[no-untyped-def]
+            yield ctx.schedule_activity("clock.first", [])
+            parent.append(ctx)
+            before = ctx.replay_now()
+            token = ambient.set("parent")
+            try:
+                outcome = replay(Discard, [], [], payload_codec="avro")
+                assert ambient.get() == "suspended"
+                if collected:
+                    cycle: list[Any] = [outcome]
+                    cycle.append(cycle)
+                    del outcome, cycle
+                    gc.collect()
+                else:
+                    outcome.close()
+                assert observations == [("suspended", None, None), "parent"]
+                assert ambient.get() == "parent"
+                assert ctx.replay_now() == before
+                assert _ACTIVE_WORKFLOW_REPLAY.get() is ctx
+            finally:
+                ambient.reset(token)
+            return clock_text(ctx)
+
+    assert result(history()[:3], Parent) == "2026-11-01T00:00:03.123456+00:00"
+    assert ambient.get() == "outside"
+    assert _ACTIVE_WORKFLOW_REPLAY.get() is None
 
 
 def _native_fixture() -> dict[str, Any]: