Advance replay time for persisted scalar signal delivery

docs/replay-clock-source.md

@@ -22,12 +22,13 @@ counter identity to order emitted events.
 | Scalar remote activity | Existing validated command sequence/type/result correlation, then the selected terminal event | Selected recorded time; successful redrive reuse retains `reused_recorded_at`; the existing Core failure branch uses its event time |
 | Scalar plain timer | Existing validated `TimerScheduled`/`TimerFired` pair, exact sequence/ID/delay | Matched fire time, with recorded reuse provenance |
 | Scalar child result | One schedule, optional start and unique parent `ChildRunCompleted` / `ChildRunFailed` / `ChildRunCancelled` / `ChildRunTerminated`, with matching nonempty child call, instance, run and type identities | Parent resolution event time for both send and throw; child `closed_at` and `reused_recorded_at` are not this branch's clock authority |
-| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable |
+| Scalar persisted signal delivery | One untimed `SignalWaitOpened` followed by a unique `SignalApplied`, matching nonempty wait ID/name and explicit command sequence, with the persisted value consumed | Applied event time, preserving valid `reused_recorded_at` exactly as Core signal resume does |
+| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child/signal correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable |
 | Replayed-instance query | Clock of the actual reconstructed workflow prefix, rather than a later visible row | Same supported boundary rules |
 | Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable |
 | Executor condition predicate | Current executor exposes a full signal/update view without retaining the exact triggering consumed prefix | Unavailable inside the predicate; enclosing workflow scope is restored on return/error |
 | Inline side effect/version/memo/search/local result | User code can resume before its event is persisted | Unavailable on first execution and the corresponding cold path in this cohort; no implicit marker/yield/fence |
-| Signal/condition result, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact |
+| Condition result, timed signal wait, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact |
 
 The mutable workflow clock is bound only during handler construction and each
 future poll. A separate execution-local guard restores its predecessor on
@@ -77,3 +78,21 @@ identity/time refusals, snapshot/reconstructed query prefixes and interleaved
 poll/error scope restoration. The separate child corpus consumer runs unchanged
 against base and candidate production code, retaining an exact behavioral
 negative rather than a compilation failure.
+
+The scalar persisted-signal cohort follows Core
+[`4b2cdc55` `WorkflowExecutor::resumeSignal`](https://github.com/durable-workflow/workflow/blob/4b2cdc5507120d9c213b0b02a85738bec96c574a/src/V2/Support/WorkflowExecutor.php#L1434),
+which supplies `RedriveReplayClock::recordedAt` for the matched `SignalApplied`.
+Rust consumes that recorded wait by command sequence and signal name. Business
+time additionally requires matching nonempty persisted wait IDs and explicit
+numeric command counters, the canonical persisted value, one ordered open/apply
+pair and no timed-wait evidence. Missing/conflicting identity, duplicate/order
+ambiguity, invalid/missing time or a timed wait leaves time unavailable while
+preserving existing signal results, commands and cancellation behavior.
+
+A visible `SignalReceived` is not application. Neither an unconsumed future row
+nor task-resume arguments without the committed applied carrier advance this
+clock. Core query reconstruction has a separate committed received-event legacy
+fallback; this cohort does not invent that authority for Rust or read a mutable
+projection as persisted time. No new observation envelope, event, wire field,
+marker, RPC or inline persistence fence is introduced. Reconstructed/snapshot
+queries retain the existing actually consumed/inspected prefix rules.

src/replay_clock.rs

@@ -1,9 +1,9 @@
 //! Additive business time, independent of cancellation's monotonic budget clock.
 
 use super::{
-    durable_event_sequence, is_internal_timer_event, recorded_commands, DurableOperationHandle,
-    Error, HistoryEvent, ParallelGroupMetadata, QueryContext, RecordedCommand, Result, Value,
-    WorkflowContext, WorkflowIdentity, WorkflowState,
+    durable_event_sequence, is_internal_timer_event, is_recorded_signal_wait_event,
+    recorded_commands, DurableOperationHandle, Error, HistoryEvent, ParallelGroupMetadata,
+    QueryContext, RecordedCommand, Result, Value, WorkflowContext, WorkflowIdentity, WorkflowState,
 };
 use chrono::{DateTime, Utc};
 use std::{
@@ -80,6 +80,46 @@ enum BoundaryKind {
     Activity,
     Timer,
     Child,
+    Signal,
+}
+
+fn signal_boundary(history: &[HistoryEvent], indices: &[usize], sequence: u64) -> Option<usize> {
+    let [opened, applied] = indices else {
+        return None;
+    };
+    let open = &history[*opened];
+    let apply = &history[*applied];
+    if open.event_type != "SignalWaitOpened"
+        || apply.event_type != "SignalApplied"
+        || open
+            .payload
+            .get("timeout_seconds")
+            .is_some_and(|value| !value.is_null())
+        || apply.payload.get("value").is_none_or(Value::is_null)
+    {
+        return None;
+    }
+    for event in [open, apply] {
+        if event.payload.get("sequence").and_then(Value::as_u64) != Some(sequence)
+            || event
+                .payload
+                .get("workflow_sequence")
+                .is_some_and(|value| value.as_u64() != Some(sequence))
+        {
+            return None;
+        }
+    }
+    for key in ["signal_wait_id", "signal_name"] {
+        let identity = open
+            .payload
+            .get(key)?
+            .as_str()
+            .filter(|value| !value.is_empty())?;
+        if apply.payload.get(key).and_then(Value::as_str) != Some(identity) {
+            return None;
+        }
+    }
+    Some(*applied)
 }
 
 fn child_boundary(history: &[HistoryEvent], indices: &[usize]) -> Option<usize> {
@@ -153,11 +193,20 @@ impl Clock {
                 } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => {
                     Some((*sequence, BoundaryKind::Child))
                 }
+                RecordedCommand::SignalWait {
+                    sequence,
+                    value: Some(_),
+                    parallel_group_path,
+                    ..
+                } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => {
+                    Some((*sequence, BoundaryKind::Signal))
+                }
                 _ => None,
             })
             .collect();
         let mut boundaries = HashMap::new();
         let mut children: HashMap<u64, Vec<usize>> = HashMap::new();
+        let mut signals: HashMap<u64, Vec<usize>> = HashMap::new();
         for (index, event) in history.iter().enumerate() {
             let Some(sequence) = durable_event_sequence(event) else {
                 continue;
@@ -181,6 +230,14 @@ impl Clock {
                     }
                     false
                 }
+                Some(BoundaryKind::Signal) => {
+                    if is_recorded_signal_wait_event(event)
+                        || (event.event_type.starts_with("Timer") && is_internal_timer_event(event))
+                    {
+                        signals.entry(sequence).or_default().push(index);
+                    }
+                    false
+                }
                 None => false,
             };
             if matched {
@@ -192,6 +249,11 @@ impl Clock {
                 boundaries.insert(sequence, index);
             }
         }
+        for (sequence, indices) in signals {
+            if let Some(index) = signal_boundary(history, &indices, sequence) {
+                boundaries.insert(sequence, index);
+            }
+        }
         // Index identities once. Timestamps are read only when that boundary
         // is actually consumed, never by scanning ahead from the accessor.
         Self {
@@ -264,7 +326,7 @@ impl WorkflowState {
                     event,
                     matches!(
                         event.event_type.as_str(),
-                        "ActivityCompleted" | "TimerFired"
+                        "ActivityCompleted" | "TimerFired" | "SignalApplied"
                     ),
                 )
             });
@@ -279,7 +341,7 @@ impl WorkflowContext {
     /// or invalid evidence returns [`Error::ReplayClockUnavailable`]. The clock
     /// is available only while this workflow's handler/future is executing.
     ///
-    /// This source cohort supports correlated scalar activity/timer/child
+    /// This source cohort supports correlated scalar activity/timer/child/signal
     /// results and the explicit redrive start anchor. Ordinary start authority,
     /// inline metadata/local completions and grouped/other operation families
     /// remain unavailable. Cancellation's existing budget clock is unchanged.
@@ -323,6 +385,7 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option<Dat
             | "ChildRunFailed"
             | "ChildRunCancelled"
             | "ChildRunTerminated"
+            | "SignalApplied"
     ) {
         return None;
     }
@@ -344,7 +407,7 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option<Dat
                 event,
                 matches!(
                     event.event_type.as_str(),
-                    "ActivityCompleted" | "TimerFired"
+                    "ActivityCompleted" | "TimerFired" | "SignalApplied"
                 ),
             )
         })

src/replay_clock/tests.rs

@@ -515,3 +515,4 @@ async fn replay_now_reconstructed_query_uses_consumed_prefix_not_snapshot_tail()
 }
 
 mod child;
+mod signal;