From 96c93b2808b8e8f35d8345c2f007698d352fd4ed Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 07:20:58 +0000
Subject: [PATCH 1/2] fix: expose replay time for parallel timer groups

---
 docs/replay-clock-source.md                   |  17 ++
 regression-corpus-policy.json                 |   3 +
 scripts/ci/test-validate-regression-corpus.py |  17 ++
 src/replay_clock.rs                           |  27 +-
 src/replay_clock/parallel_activity.rs         |  12 +-
 src/replay_clock/parallel_timer.rs            | 143 ++++++++++
 src/replay_clock/tests.rs                     |   2 +
 src/replay_clock/tests/parallel_timer.rs      | 259 ++++++++++++++++++
 .../business-replay-clock-parallel-timer.json | 140 ++++++++++
 tests/replay_regression_corpus.rs             |   4 +
 .../business_clock.rs                         |   9 +
 11 files changed, 618 insertions(+), 15 deletions(-)
 create mode 100644 src/replay_clock/parallel_timer.rs
 create mode 100644 src/replay_clock/tests/parallel_timer.rs
 create mode 100644 tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json

diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md
index ec97d86..0b65180 100644
--- a/docs/replay-clock-source.md
+++ b/docs/replay-clock-source.md
@@ -26,6 +26,7 @@ counter identity to order emitted events.
 | Scalar persisted condition outcome | One ordered canonical wait open/resolution, matching physical wait, logical occurrence, explicit command counter, key/fingerprint and timeout definition | Consumed satisfied/timed-out resolution event time, preserving its precision; timer and scalar redrive helpers are not substituted |
 | Persisted remote Activity all group | Exact authored group/path and each consumed member lifecycle/type/sequence/execution identity | Core millisecond aggregate on success, full chosen precision; exact agreed failure at Core millisecond precision |
 | Persisted scalar condition timeout without explicit resolution | One canonical wait and its exact ordered schedule/fire pair, with complete identity/definition/counters and unique physical timer | Executor uses the fired event’s own `recorded_at`; Core query has no corresponding resolution authority |
+| Persisted plain Timer all group | Exact authored paths/counters and one matching schedule/fire per member, with unique physical timer IDs | Core consumed-group millisecond fold, first descriptor on ties, full selected event precision; no grouped query authority |
 | Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child/signal/condition 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 scalar boundary rules; grouped executor aggregation is not query authority |
 | Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable |
@@ -193,3 +194,19 @@ preserved. Grouped conditions, predicate-trigger retention and the separate PHP
 occurrence-carrier gap remain open, as do published/native conformance and
 release adoption. The corpus fixture is synthetic source evidence shaped from
 the immutable writer, not a claim that a native experiment ran.
+
+Persisted plain Timer `all` follows Core
+[`dispatchParallelTimer`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L2007).
+Each authored descriptor needs one ordered schedule/fire, matching nonempty timer
+ID, numeric command counters, delay and complete group path. Internal or unknown
+timer kinds, cancellation, duplicates, aliases or unsupported time evidence do
+not supply an aggregate. Nested all-timer paths preserve descriptor order.
+
+The fold uses the same measured Carbon comparator as Activity groups. It selects
+only consumed group members, keeps the first descriptor on a rounded-ms tie,
+and returns that member's full represented precision. Raw `recorded_at` or Core's
+represented `created_at` fallback supplies time; scalar reuse helpers and payload
+fired/deadline timestamps do not. Pending/empty behavior and grouped query
+unavailability retain the existing rules. Mixed groups and other group families
+remain outside this bounded source mapping. No marker, wire or producer change
+is introduced, and no installed/native or published conformance is inferred.
diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json
index b46ba91..5b23721 100644
--- a/regression-corpus-policy.json
+++ b/regression-corpus-policy.json
@@ -80,6 +80,9 @@
         },
         {
           "glob": "src/replay_clock/condition_timeout.rs"
+        },
+        {
+          "glob": "src/replay_clock/parallel_timer.rs"
         }
       ]
     }
diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py
index c77d21f..2a5a06a 100644
--- a/scripts/ci/test-validate-regression-corpus.py
+++ b/scripts/ci/test-validate-regression-corpus.py
@@ -697,6 +697,23 @@ def matches(guard: dict[str, object]) -> bool:
             for guard in guards
         ))
 
+    def test_parallel_timer_clock_module_requires_replay_evidence(self) -> None:
+        source = "src/replay_clock/parallel_timer.rs"
+        path = self.root / source
+        path.parent.mkdir(parents=True)
+        path.write_text("fn boundary() -> Option<usize> { Some(0) }\n", encoding="utf-8")
+        guards = REPOSITORY_POLICY["categories"]["replay"]["guards"]
+
+        def matches(guard: dict[str, object]) -> bool:
+            return VALIDATOR._guard_matches(self.root, "HEAD", {source}, guard)
+
+        self.assertFalse(any(matches(guard) for guard in guards if guard["glob"] != source))
+        self.assertTrue(any(matches(guard) for guard in guards))
+        self.assertFalse(any(
+            VALIDATOR._guard_matches(self.root, "HEAD", {"src/replay_clock/tests/parallel_timer.rs"}, guard)
+            for guard in guards
+        ))
+
     def test_history_fixture_in_unchanged_test_module_is_not_production_replay(self) -> None:
         diff = """\
 diff --git a/src/lib.rs b/src/lib.rs
diff --git a/src/replay_clock.rs b/src/replay_clock.rs
index 1910428..1c2f343 100644
--- a/src/replay_clock.rs
+++ b/src/replay_clock.rs
@@ -15,6 +15,7 @@ use std::{
 mod condition;
 mod condition_timeout;
 mod parallel_activity;
+mod parallel_timer;
 
 thread_local! {
     static ACTIVE: RefCell<Option<Weak<Mutex<WorkflowState>>>> = const { RefCell::new(None) };
@@ -112,6 +113,7 @@ pub(super) struct Clock {
     time: Option<DateTime<Utc>>,
     boundaries: HashMap<u64, usize>,
     parallel_activities: parallel_activity::Members,
+    parallel_timers: parallel_timer::Members,
 }
 
 enum BoundaryKind {
@@ -405,6 +407,7 @@ impl Clock {
             time: start_time(history),
             boundaries,
             parallel_activities: parallel_activity::Members::new(history, commands),
+            parallel_timers: parallel_timer::Members::new(history, commands),
         }
     }
 
@@ -438,11 +441,16 @@ impl WorkflowState {
         self.replay_clock.time = if QUERY_REPLAY.with(Cell::get) {
             None
         } else {
-            self.replay_clock.parallel_activities.consumed_time(
-                &self.history_events,
-                sequences,
-                failed,
-            )
+            self.replay_clock
+                .parallel_activities
+                .consumed_time(&self.history_events, sequences, failed)
+                .or_else(|| {
+                    self.replay_clock.parallel_timers.consumed_time(
+                        &self.history_events,
+                        sequences,
+                        failed,
+                    )
+                })
         };
         self.advance_cancellation_sequences(sequences, failed);
     }
@@ -489,10 +497,11 @@ 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/signal
-    /// results, persisted Activity all groups, and the explicit redrive start anchor. Ordinary start authority,
-    /// inline metadata/local completions and other grouped operation families
-    /// remain unavailable. Cancellation's existing budget clock is unchanged.
+    /// Supported boundaries are correlated scalar activity, timer, child, signal
+    /// and condition results, persisted Activity and Timer all groups, and the
+    /// explicit redrive start anchor. Ordinary start authority, inline metadata
+    /// and local completions, and other grouped families remain unavailable.
+    /// Cancellation's existing budget clock is unchanged.
     /// Migrate business decisions through normal workflow versioning. Event
     /// timestamps do not provide a total order; retain stream/counter identity.
     pub fn replay_now(&self) -> Result<DateTime<Utc>> {
diff --git a/src/replay_clock/parallel_activity.rs b/src/replay_clock/parallel_activity.rs
index 51b551a..b841825 100644
--- a/src/replay_clock/parallel_activity.rs
+++ b/src/replay_clock/parallel_activity.rs
@@ -15,10 +15,10 @@ struct Member {
 #[derive(Debug)]
 pub(super) struct Members(HashMap<u64, Member>);
 
-fn all_activity(path: &[ParallelGroupMetadata]) -> bool {
+pub(super) fn all_path(path: &[ParallelGroupMetadata], kind: &str) -> bool {
     !path.is_empty()
         && path.iter().all(|entry| {
-            entry.parallel_group_kind == "activity"
+            entry.parallel_group_kind == kind
                 && entry
                     .parallel_group_mode
                     .as_deref()
@@ -110,7 +110,7 @@ pub(super) fn milliseconds(time: DateTime<Utc>) -> Option<i64> {
     Some((micros as f64 / 1000.0).round() as i64)
 }
 
-fn time(event: &HistoryEvent) -> Option<DateTime<Utc>> {
+pub(super) fn event_time(event: &HistoryEvent) -> Option<DateTime<Utc>> {
     match event.raw.get("recorded_at") {
         None | Some(Value::Null) => parse(event.raw.get("created_at")),
         present => parse(present),
@@ -139,7 +139,7 @@ impl Members {
                     else {
                         return None;
                     };
-                    if !all_activity(path) {
+                    if !all_path(path, "activity") {
                         return None;
                     }
                     member(history, events.get(sequence)?, *sequence, path)
@@ -178,13 +178,13 @@ impl Members {
                 return None;
             }
             if member.failed {
-                let ms = milliseconds(time(history.get(member.terminal?)?)?)?;
+                let ms = milliseconds(event_time(history.get(member.terminal?)?)?)?;
                 // Descriptor iteration retains the first member on a tie.
                 if failure.is_none_or(|(current, _)| ms < current) {
                     failure = Some((ms, *sequence));
                 }
             } else if failed.is_none() {
-                let time = time(history.get(member.terminal?)?)?;
+                let time = event_time(history.get(member.terminal?)?)?;
                 let ms = milliseconds(time)?;
                 if success.is_none_or(|(current, _)| ms > current) {
                     success = Some((ms, time));
diff --git a/src/replay_clock/parallel_timer.rs b/src/replay_clock/parallel_timer.rs
new file mode 100644
index 0000000..e696816
--- /dev/null
+++ b/src/replay_clock/parallel_timer.rs
@@ -0,0 +1,143 @@
+use super::parallel_activity::{all_path, event_time, milliseconds};
+use super::{durable_event_sequence, DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value};
+use crate::{recorded_parallel_group_path, ParallelGroupMetadata};
+use std::collections::HashSet;
+
+#[derive(Debug)]
+struct Member {
+    path: Vec<ParallelGroupMetadata>,
+    timer: String,
+    fired: usize,
+}
+
+#[derive(Debug)]
+pub(super) struct Members(HashMap<u64, Member>);
+
+fn member(
+    history: &[HistoryEvent],
+    indices: &[usize],
+    sequence: u64,
+    delay: u64,
+    path: &[ParallelGroupMetadata],
+) -> Option<Member> {
+    let [scheduled, fired] = indices else {
+        return None;
+    };
+    if history[*scheduled].event_type != "TimerScheduled"
+        || history[*fired].event_type != "TimerFired"
+    {
+        return None;
+    }
+    let timer = history[*scheduled]
+        .payload
+        .get("timer_id")?
+        .as_str()
+        .filter(|value| !value.is_empty())?;
+    for &index in indices {
+        let event = &history[index];
+        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))
+            || event.payload.get("timer_id").and_then(Value::as_str) != Some(timer)
+            || event.payload.get("delay_seconds").and_then(Value::as_u64) != Some(delay)
+            || event
+                .payload
+                .get("timer_kind")
+                .is_some_and(|value| !value.is_null())
+            || crate::is_internal_timer_event(event)
+            || recorded_parallel_group_path(&[event], sequence)
+                .ok()
+                .flatten()
+                .as_deref()
+                != Some(path)
+        {
+            return None;
+        }
+    }
+    Some(Member {
+        path: path.to_vec(),
+        timer: timer.into(),
+        fired: *fired,
+    })
+}
+
+impl Members {
+    pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self {
+        let mut events: HashMap<u64, Vec<usize>> = HashMap::new();
+        for (index, event) in history.iter().enumerate() {
+            if event.event_type.starts_with("Timer") {
+                if let Some(sequence) = durable_event_sequence(event) {
+                    events.entry(sequence).or_default().push(index);
+                }
+            }
+        }
+        Self(
+            commands
+                .iter()
+                .filter_map(|command| {
+                    let RecordedCommand::Timer {
+                        sequence,
+                        delay_seconds,
+                        fired: true,
+                        parallel_group_path: Some(path),
+                    } = command
+                    else {
+                        return None;
+                    };
+                    if !all_path(path, "timer") {
+                        return None;
+                    }
+                    member(
+                        history,
+                        events.get(sequence)?,
+                        *sequence,
+                        *delay_seconds,
+                        path,
+                    )
+                    .map(|member| (*sequence, member))
+                })
+                .collect(),
+        )
+    }
+
+    pub(super) fn consumed_time(
+        &self,
+        history: &[HistoryEvent],
+        sequences: &[u64],
+        failed: Option<u64>,
+    ) -> Option<DateTime<Utc>> {
+        if failed.is_some() {
+            return None;
+        }
+        let root = self.0.get(sequences.first()?)?.path.first()?;
+        if root.parallel_group_base_sequence != *sequences.first()?
+            || root.parallel_group_size != sequences.len()
+        {
+            return None;
+        }
+        let mut timers = HashSet::new();
+        let mut selected: Option<(i64, DateTime<Utc>)> = None;
+        for (index, sequence) in sequences.iter().enumerate() {
+            let member = self.0.get(sequence)?;
+            let mut expected = root.clone();
+            expected.parallel_group_index = index;
+            if member.path.first()? != &expected
+                || root
+                    .parallel_group_base_sequence
+                    .checked_add(index as u64)?
+                    != *sequence
+                || !timers.insert(&member.timer)
+            {
+                return None;
+            }
+            let time = event_time(history.get(member.fired)?)?;
+            let ms = milliseconds(time)?;
+            if selected.is_none_or(|(current, _)| ms > current) {
+                selected = Some((ms, time));
+            }
+        }
+        selected.map(|(_, time)| time)
+    }
+}
diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs
index fe88fc2..0ac9d1d 100644
--- a/src/replay_clock/tests.rs
+++ b/src/replay_clock/tests.rs
@@ -522,3 +522,5 @@ mod condition;
 mod parallel_activity;
 
 mod condition_timeout;
+
+mod parallel_timer;
diff --git a/src/replay_clock/tests/parallel_timer.rs b/src/replay_clock/tests/parallel_timer.rs
new file mode 100644
index 0000000..9e9a50a
--- /dev/null
+++ b/src/replay_clock/tests/parallel_timer.rs
@@ -0,0 +1,259 @@
+use super::*;
+use crate::{ParallelOperation, ParallelResult};
+
+fn operations(nested: bool) -> Vec<ParallelOperation> {
+    let timer = || ParallelOperation::timer(Duration::from_secs(1));
+    if nested {
+        vec![timer(), ParallelOperation::group(vec![timer()])]
+    } else {
+        vec![timer(), timer()]
+    }
+}
+
+fn group(base: u64, times: [&str; 2], nested: bool) -> Vec<HistoryEvent> {
+    let descriptors = crate::parallel_descriptors(operations(nested), base).unwrap();
+    let mut schedules = Vec::new();
+    let mut fires = Vec::new();
+    for (index, descriptor) in descriptors.iter().enumerate() {
+        let sequence = base + index as u64;
+        let mut payload = json!({"sequence":sequence,"timer_id":format!("timer-{sequence}"),"delay_seconds":1,"fire_at":"2099-01-01T00:00:00Z"});
+        crate::apply_parallel_group_path(payload.as_object_mut().unwrap(), &descriptor.group_path);
+        schedules.push(event("TimerScheduled", payload.clone(), START));
+        payload["fired_at"] = json!("2099-01-01T00:00:01Z");
+        fires.push(event("TimerFired", payload, times[index]));
+    }
+    schedules.extend(fires);
+    schedules
+}
+
+fn observe(history: Vec<HistoryEvent>, nested: bool) -> Value {
+    let ctx = context(history);
+    let inner = ctx.clone();
+    let output = ready(
+        &ctx,
+        Box::pin(async move {
+            let values = inner.parallel(operations(nested)).await?;
+            assert_eq!(values[0], ParallelResult::Timer);
+            assert_eq!(
+                values[1],
+                if nested {
+                    ParallelResult::Group(vec![ParallelResult::Timer])
+                } else {
+                    ParallelResult::Timer
+                }
+            );
+            let time = inner.replay_now().ok();
+            assert_eq!(time, inner.replay_now().ok());
+            AvroValue::from_serialize(&json!(
+                time.map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true))
+            ))
+        }),
+    )
+    .into_json()
+    .unwrap();
+    assert!(ctx.take_commands().unwrap().is_empty());
+    assert!(ctx.replay_now().is_err());
+    output
+}
+
+#[test]
+fn replay_now_timer_all_uses_descriptor_order_core_ms_ties_and_full_precision_on_cold_replay() {
+    for nested in [false, true] {
+        for (times, expected) in [
+            (
+                ["2026-01-01T00:00:05.123100Z", "2026-01-01T00:00:05.123400Z"],
+                "2026-01-01T00:00:05.123100Z",
+            ),
+            (
+                ["2026-01-01T00:00:05.123499Z", "2026-01-01T00:00:05.123500Z"],
+                "2026-01-01T00:00:05.123500Z",
+            ),
+            (
+                ["1969-12-31T23:59:59.999500Z", "1969-12-31T23:59:59.999501Z"],
+                "1969-12-31T23:59:59.999501Z",
+            ),
+            (
+                ["2025-12-31T23:59:59.654321Z", "2025-12-31T23:59:58.654321Z"],
+                "2025-12-31T23:59:59.654321Z",
+            ),
+        ] {
+            let mut history = vec![start()];
+            history.extend(group(1, times, nested));
+            history.swap(3, 4);
+            assert_eq!(observe(history.clone(), nested), json!(expected));
+            assert_eq!(observe(history.clone(), nested), observe(history, nested));
+        }
+    }
+}
+
+#[test]
+fn replay_now_timer_all_never_borrows_reuse_payload_or_other_groups_future_times() {
+    let mut history = group(1, [START; 2], false);
+    for event in &mut history[2..] {
+        event.payload["reused_recorded_at"] = json!("2099-01-01T00:00:00Z");
+        event
+            .raw
+            .insert("timestamp".into(), json!("2099-01-01T00:00:00Z"));
+    }
+    history.extend(group(3, ["invalid-future-time"; 2], false));
+    assert_eq!(observe(history, false), json!(START));
+}
+
+#[test]
+fn replay_now_timer_all_missing_or_invalid_time_has_no_scalar_fallback() {
+    for value in [None, Some(Value::Null), Some(json!("bad")), Some(json!(42))] {
+        let mut history = group(1, [START; 2], false);
+        match value {
+            Some(value) => {
+                history[2].raw.insert("recorded_at".into(), value);
+            }
+            None => {
+                history[2].raw.remove("recorded_at");
+            }
+        }
+        history[2].payload["reused_recorded_at"] = json!(START);
+        assert!(observe(history, false).is_null());
+    }
+    let mut history = group(1, [START; 2], false);
+    history[2].raw.remove("recorded_at");
+    history[2]
+        .raw
+        .insert("created_at".into(), json!("2026-01-01T00:00:10.654321Z"));
+    assert_eq!(
+        observe(history, false),
+        json!("2026-01-01T00:00:10.654321Z")
+    );
+    for time in ["2500-01-01T00:00:00Z", "2026-01-01T00:00:00.123456789Z"] {
+        assert!(observe(group(1, [time; 2], false), false).is_null());
+    }
+}
+
+#[test]
+fn replay_now_timer_all_requires_typed_counter_unique_identity_and_complete_plain_timer_evidence() {
+    let original = group(1, [START; 2], false);
+    for index in 0..4 {
+        let sequence = if index % 2 == 0 { 1 } else { 2 };
+        let mut history = original.clone();
+        history[index].payload["sequence"] = json!(sequence.to_string());
+        assert!(observe(history, false).is_null());
+        let mut history = original.clone();
+        history[index]
+            .payload
+            .as_object_mut()
+            .unwrap()
+            .remove("sequence");
+        history[index]
+            .raw
+            .insert("sequence".into(), json!(sequence));
+        assert!(observe(history, false).is_null());
+        let mut history = original.clone();
+        history[index].payload["workflow_sequence"] = json!(99);
+        assert!(observe(history, false).is_null());
+        let mut history = original.clone();
+        history[index].payload["timer_kind"] = json!("unknown-kind");
+        assert!(observe(history, false).is_null());
+    }
+    let mut history = original.clone();
+    for event in &mut history {
+        event.payload["timer_id"] = json!("one-aliased-timer");
+    }
+    assert!(observe(history, false).is_null());
+    let mut history = original.clone();
+    let mut cancel = history[2].clone();
+    cancel.event_type = "TimerCancelled".into();
+    history.push(cancel);
+    assert!(observe(history, false).is_null());
+    let mut history = original;
+    history.swap(0, 2);
+    assert!(observe(history, false).is_null());
+}
+
+#[test]
+fn replay_now_timer_all_keeps_legacy_duplicate_and_mismatched_timer_refusals() {
+    for change in ["duplicate", "identity", "delay"] {
+        let mut history = group(1, [START; 2], false);
+        match change {
+            "duplicate" => history.push(history[2].clone()),
+            "identity" => history[2].payload["timer_id"] = json!("foreign"),
+            "delay" => history[2].payload["delay_seconds"] = json!(2),
+            _ => unreachable!(),
+        }
+        assert!(matches!(
+            WorkflowState::new(history, "clock-tests".into(), DEFAULT_CODEC.into(), None),
+            Err(Error::NonDeterministicReplay(_))
+        ));
+    }
+}
+
+#[test]
+fn replay_now_pending_timer_group_has_no_aggregate_and_empty_group_keeps_previous_prefix() {
+    let mut history = group(1, [START; 2], false);
+    history.pop();
+    let ctx = context(history);
+    let inner = ctx.clone();
+    let mut future: WorkflowFuture = Box::pin(async move {
+        inner.parallel(operations(false)).await?;
+        Ok(AvroValue::Null)
+    });
+    assert!(Worker::poll_workflow_future(&ctx, &mut future)
+        .unwrap()
+        .is_pending());
+    {
+        let _guard = Guard::enter(&ctx);
+        assert!(ctx.replay_now().is_err());
+    }
+    let ctx = context(vec![start()]);
+    let inner = ctx.clone();
+    ready(
+        &ctx,
+        Box::pin(async move {
+            inner.parallel(vec![]).await?;
+            assert_eq!(read(&inner), START);
+            Ok(AvroValue::Null)
+        }),
+    );
+}
+
+#[tokio::test]
+async fn replay_now_timer_groups_do_not_supply_snapshot_or_reconstructed_query_time() {
+    let mut worker = Worker::new(Client::new("http://127.0.0.1:8080").unwrap(), "clock-tests");
+    worker.register_replayed_workflow(
+        "timer-group-clock",
+        || 0_u64,
+        |ctx, _, _| async move {
+            ctx.parallel(operations(false)).await?;
+            assert!(ctx.replay_now().is_err());
+            ctx.sleep(Duration::from_secs(1)).await?;
+            Ok(Value::Null)
+        },
+    );
+    worker.register_query("timer-group-clock", "snapshot", |ctx, _| async move {
+        Ok(json!(ctx.replay_now().is_err()))
+    });
+    worker.register_replayed_query::<u64, _, _>(
+        "timer-group-clock",
+        "reconstructed",
+        |ctx, _, _| async move { Ok(json!(ctx.replay_now().is_err())) },
+    );
+    let mut history = group(1, [START; 2], false);
+    for query in ["snapshot", "reconstructed"] {
+        if query == "reconstructed" {
+            history.push(event(
+                "TimerScheduled",
+                json!({"sequence":3,"timer_id":"later","delay_seconds":1}),
+                "2099-01-01T00:00:00Z",
+            ));
+        }
+        let task:QueryTask=serde_json::from_value(json!({"query_task_id":"timer-group-q","workflow_type":"timer-group-clock","query_name":query,"payload_codec":DEFAULT_CODEC,"history_events":history})).unwrap();
+        assert_eq!(
+            worker
+                .execute_query_task(task)
+                .await
+                .unwrap()
+                .into_json()
+                .unwrap(),
+            json!(true)
+        );
+    }
+    assert_eq!(observe(group(1, [START; 2], false), false), json!(START));
+}
diff --git a/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json b/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json
new file mode 100644
index 0000000..fd9454c
--- /dev/null
+++ b/tests/fixtures/replay-regressions/business-replay-clock-parallel-timer.json
@@ -0,0 +1,140 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "rust-business-replay-clock-parallel-timer",
+  "protocol_version": "1.20",
+  "bindings": [
+    "rust"
+  ],
+  "workflow": {
+    "type": "corpus.business-replay-clock-parallel-timer",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 1,
+        "timer_id": "clock-timer-1",
+        "delay_seconds": 1,
+        "fire_at": "2026-01-01T00:00:01Z",
+        "parallel_group_id": "parallel-timers:1:2",
+        "parallel_group_kind": "timer",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-timers:1:2",
+            "parallel_group_kind": "timer",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 1
+    },
+    {
+      "event_type": "TimerScheduled",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "clock-timer-2",
+        "delay_seconds": 1,
+        "fire_at": "2026-01-01T00:00:01Z",
+        "parallel_group_id": "parallel-timers:1:2",
+        "parallel_group_kind": "timer",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-timers:1:2",
+            "parallel_group_kind": "timer",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 2
+    },
+    {
+      "event_type": "TimerFired",
+      "payload": {
+        "sequence": 1,
+        "timer_id": "clock-timer-1",
+        "delay_seconds": 1,
+        "fire_at": "2026-01-01T00:00:01Z",
+        "parallel_group_id": "parallel-timers:1:2",
+        "parallel_group_kind": "timer",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-timers:1:2",
+            "parallel_group_kind": "timer",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:05.123100Z",
+      "sequence": 3
+    },
+    {
+      "event_type": "TimerFired",
+      "payload": {
+        "sequence": 2,
+        "timer_id": "clock-timer-2",
+        "delay_seconds": 1,
+        "fire_at": "2026-01-01T00:00:01Z",
+        "parallel_group_id": "parallel-timers:1:2",
+        "parallel_group_kind": "timer",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-timers:1:2",
+            "parallel_group_kind": "timer",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:05.123400Z",
+      "sequence": 4
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "entry_unavailable": true,
+      "members": 2,
+      "time": "2026-01-01T00:00:05.123100Z",
+      "unchanged_prefix": true
+    }
+  },
+  "command_sequence": [
+    {
+      "type": "complete_workflow",
+      "result": {
+        "entry_unavailable": true,
+        "members": 2,
+        "time": "2026-01-01T00:00:05.123100Z",
+        "unchanged_prefix": true
+      }
+    }
+  ],
+  "worker_task": {
+    "workflow_id": "timer-group-clock-parent",
+    "payload_codec": "avro"
+  }
+}
diff --git a/tests/replay_regression_corpus.rs b/tests/replay_regression_corpus.rs
index 1e142ac..8af64a6 100644
--- a/tests/replay_regression_corpus.rs
+++ b/tests/replay_regression_corpus.rs
@@ -511,6 +511,7 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result<
             | "corpus.business-replay-clock-condition"
             | "corpus.business-replay-clock-condition-timeout"
             | "corpus.business-replay-clock-parallel-activity"
+            | "corpus.business-replay-clock-parallel-timer"
             | "corpus.nexus-result"
             | "corpus.nexus-cancellation"
     ) {
@@ -630,6 +631,9 @@ async fn execute_fixture_delivery(fixture: &Value, delivery_id: &str) -> Result<
         "corpus.business-replay-clock-signal" => {
             business_clock::register_signal(&mut worker, workflow_type)
         }
+        "corpus.business-replay-clock-parallel-timer" => {
+            business_clock::register_parallel_timer(&mut worker, workflow_type)
+        }
         "corpus.business-replay-clock-parallel-activity" => {
             business_clock::register_parallel_activity(&mut worker, workflow_type)
         }
diff --git a/tests/replay_regression_corpus/business_clock.rs b/tests/replay_regression_corpus/business_clock.rs
index 76afe3e..40833a5 100644
--- a/tests/replay_regression_corpus/business_clock.rs
+++ b/tests/replay_regression_corpus/business_clock.rs
@@ -159,3 +159,12 @@ pub(super) fn register_condition_timeout(worker: &mut Worker, workflow_type: &st
         }))
     });
 }
+
+pub(super) fn register_parallel_timer(worker: &mut Worker, workflow_type: &str) {
+    worker.register_workflow(workflow_type, |ctx, _| async move {
+        let entry_unavailable=ctx.replay_now().is_err();
+        let values=ctx.parallel(vec![ParallelOperation::timer(Duration::from_secs(1)),ParallelOperation::timer(Duration::from_secs(1))]).await?;
+        let time=ctx.replay_now().ok();
+        Ok(json!({"entry_unavailable":entry_unavailable,"members":values.len(),"time":time.map(|time|time.to_rfc3339_opts(SecondsFormat::Micros,true)),"unchanged_prefix":time.is_some() && time==ctx.replay_now().ok()}))
+    });
+}

From 133423875d483ffd353ad88187927a31fd455159 Mon Sep 17 00:00:00 2001
From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com>
Date: Fri, 6 Nov 2026 07:44:35 +0000
Subject: [PATCH 2/2] fix: use exact canonical replay-clock quantization

---
 docs/replay-clock-source.md                   |  48 +++-
 regression-corpus-policy.json                 |   3 +
 scripts/ci/test-validate-regression-corpus.py |  17 ++
 src/replay_clock.rs                           |   1 +
 src/replay_clock/parallel.rs                  |  59 ++++
 src/replay_clock/parallel_activity.rs         |  54 +---
 src/replay_clock/parallel_timer.rs            |  13 +-
 src/replay_clock/tests.rs                     |   2 +
 src/replay_clock/tests/parallel.rs            |  52 ++++
 src/replay_clock/tests/parallel_activity.rs   |   6 +-
 src/replay_clock/tests/parallel_timer.rs      |   2 +-
 tests/fixtures/replay-clock-canonical-v1.json | 189 +++++++++++++
 ...eplay-clock-canonical-v1-large-epochs.json | 251 ++++++++++++++++++
 13 files changed, 624 insertions(+), 73 deletions(-)
 create mode 100644 src/replay_clock/parallel.rs
 create mode 100644 src/replay_clock/tests/parallel.rs
 create mode 100644 tests/fixtures/replay-clock-canonical-v1.json
 create mode 100644 tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json

diff --git a/docs/replay-clock-source.md b/docs/replay-clock-source.md
index 0b65180..a952dad 100644
--- a/docs/replay-clock-source.md
+++ b/docs/replay-clock-source.md
@@ -24,9 +24,9 @@ counter identity to order emitted events.
 | 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 |
 | Scalar persisted signal delivery | One `SignalWaitOpened` followed by a unique `SignalApplied`, matching nonempty wait ID/name and explicit command sequence, with the persisted value consumed. A configured timeout requires valid numeric metadata and either no scheduled timer or the exact ordered schedule/cancel pair | Applied event time, preserving valid `reused_recorded_at` exactly as Core signal resume does |
 | Scalar persisted condition outcome | One ordered canonical wait open/resolution, matching physical wait, logical occurrence, explicit command counter, key/fingerprint and timeout definition | Consumed satisfied/timed-out resolution event time, preserving its precision; timer and scalar redrive helpers are not substituted |
-| Persisted remote Activity all group | Exact authored group/path and each consumed member lifecycle/type/sequence/execution identity | Core millisecond aggregate on success, full chosen precision; exact agreed failure at Core millisecond precision |
+| Persisted remote Activity all group | Exact authored group/path and each consumed member lifecycle/type/sequence/execution identity | Canonical v1 millisecond aggregate on success, full chosen precision; exact agreed failure at v1 millisecond precision |
 | Persisted scalar condition timeout without explicit resolution | One canonical wait and its exact ordered schedule/fire pair, with complete identity/definition/counters and unique physical timer | Executor uses the fired event’s own `recorded_at`; Core query has no corresponding resolution authority |
-| Persisted plain Timer all group | Exact authored paths/counters and one matching schedule/fire per member, with unique physical timer IDs | Core consumed-group millisecond fold, first descriptor on ties, full selected event precision; no grouped query authority |
+| Persisted plain Timer all group | Exact authored paths/counters and one matching schedule/fire per member, with unique physical timer IDs | Canonical v1 consumed-group millisecond fold, first descriptor on ties, full selected event precision; no grouped query authority |
 | Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer/child/signal/condition 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 scalar boundary rules; grouped executor aggregation is not query authority |
 | Direct update handler | Unique committed `UpdateAccepted` correlated to the handler's update ID | That event's validated time; missing/ambiguous acceptance is unavailable |
@@ -142,7 +142,7 @@ execution identity and complete matching group path. Nested all-activity paths
 retain authored descriptor order. Duplicate or ambiguous evidence preserves
 legacy results but cannot supply time.
 
-Success folds members in that order, comparing Carbon milliseconds and replacing
+Success folds members in that order, comparing canonical v1 milliseconds and replacing
 only on a strictly greater value. The first descriptor wins an equal-millisecond
 tie, retaining its full recorded microsecond precision. Each consumed success
 must have represented `recorded_at` or Core's actual `created_at` fallback.
@@ -151,8 +151,8 @@ unrepresented activity row never supply authority. Empty groups add no boundary;
 pending groups and other operation families have no aggregate clock here.
 
 Failure preserves Rust's existing lowest-member exception. Time is available
-only when that member also wins Core's earliest-millisecond/member-index rule;
-it is reconstructed with Core's millisecond precision. Conflicting multifailure
+only when that member also wins the canonical v1 earliest-millisecond/member-index rule;
+it is reconstructed with canonical v1 millisecond precision. Conflicting multifailure
 selection remains the [OPEN compatibility/versioning proposal](https://github.com/durable-workflow/.github/issues/215#issuecomment-6677866585),
 without rewriting legacy outcomes. Grouped snapshot and reconstructed-query time
 remain unavailable: Core's successful query group-send supplies no aggregate.
@@ -161,12 +161,12 @@ A later consumed supported scalar boundary can establish query time again.
 The measured reference in `tests/fixtures/replay-clock-carbon-reference.json`
 uses PHP 8.4.27 and Carbon 3.15.1 at `ce0e5636`. Its
 `Timestamp.php` SHA-256 is `57b2a235656f8edad7c333087a64e45491adaaf1b01ac9c40e53185b73db356d`.
-`getTimestampMs()` rounds to nearest milliseconds, with halfway values away from
-zero, including negative epochs. `createFromTimestampMs()` reconstructs those
-milliseconds. The clock accepts persisted microseconds within the exact float
-microsecond range used by that comparator; finer or out-of-range evidence stays
-unavailable. Reference vectors cover positive/negative epochs, 499/500/501
-microsecond edges, equal-millisecond candidates and timezone-equivalent instants.
+These are attributed legacy PHP 8.4/Carbon measurements, not a portable numeric
+contract. The same Carbon version can produce different rounding results under
+other PHP runtimes. Their original inputs and outputs remain unchanged.
+The separately labeled `tests/fixtures/replay-clock-canonical-v1.json` defines
+the new accessor's exact integer rule. Historical source qualifications are
+retained with their stated scope; they are not universal Core parity evidence.
 
 The persisted fired-only condition-timeout cohort follows Core
 [`resumeTimedOutCondition`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L895).
@@ -202,7 +202,7 @@ ID, numeric command counters, delay and complete group path. Internal or unknown
 timer kinds, cancellation, duplicates, aliases or unsupported time evidence do
 not supply an aggregate. Nested all-timer paths preserve descriptor order.
 
-The fold uses the same measured Carbon comparator as Activity groups. It selects
+The fold uses the same canonical v1 integer comparator as Activity groups. It selects
 only consumed group members, keeps the first descriptor on a rounded-ms tie,
 and returns that member's full represented precision. Raw `recorded_at` or Core's
 represented `created_at` fallback supplies time; scalar reuse helpers and payload
@@ -210,3 +210,27 @@ fired/deadline timestamps do not. Pending/empty behavior and grouped query
 unavailability retain the existing rules. Mixed groups and other group families
 remain outside this bounded source mapping. No marker, wire or producer change
 is introduced, and no installed/native or published conformance is inferred.
+
+The unreleased accessor's [canonical v1 numeric decision](https://github.com/durable-workflow/.github/issues/215#issuecomment-6679568700)
+uses an authoritative instant's exact signed UTC microsecond count `U`:
+`M = sign(U) * ((abs(U) + 500) div 1000)`. All operations are checked integers;
+there is no float conversion or runtime `round()`. A half millisecond rounds
+away from zero on both sides of the epoch. Unsupported precision or overflow
+is unavailable. This removes the earlier source implementation's float-range
+restriction for valid representable instants, without changing scalar event
+times or rewriting legacy group results, selected errors, cancellation or history.
+
+Success still compares only validated consumed authored members, preserves the
+first member on an equal `M`, and returns that member's original full-precision
+instant. Qualified failure reconstruction uses the selected integer `M` only
+when it agrees with the unchanged legacy winner. The multi-failure policy gap
+remains OPEN. The canonical table covers signed boundaries, offsets and large
+representable dates; a new real worker fixture independently contrasts the old
+source's unavailable large epochs against the exact v1 result.
+
+Core remains the identity/boundary/order reference. Existing Core legacy
+`Workflow::now()` values are unchanged and are not claimed numerically equivalent
+on every supported PHP runtime. A matching explicitly versioned additive-clock
+source/context on Core and published cross-SDK conformance remain required.
+Adoption uses ordinary workflow versioning. No automatic marker, run mutation,
+new public API name, live wire advertisement or release is introduced here.
diff --git a/regression-corpus-policy.json b/regression-corpus-policy.json
index 5b23721..535993e 100644
--- a/regression-corpus-policy.json
+++ b/regression-corpus-policy.json
@@ -83,6 +83,9 @@
         },
         {
           "glob": "src/replay_clock/parallel_timer.rs"
+        },
+        {
+          "glob": "src/replay_clock/parallel.rs"
         }
       ]
     }
diff --git a/scripts/ci/test-validate-regression-corpus.py b/scripts/ci/test-validate-regression-corpus.py
index 2a5a06a..283d96f 100644
--- a/scripts/ci/test-validate-regression-corpus.py
+++ b/scripts/ci/test-validate-regression-corpus.py
@@ -714,6 +714,23 @@ def matches(guard: dict[str, object]) -> bool:
             for guard in guards
         ))
 
+    def test_parallel_common_clock_module_requires_replay_evidence(self) -> None:
+        source = "src/replay_clock/parallel.rs"
+        path = self.root / source
+        path.parent.mkdir(parents=True)
+        path.write_text("fn boundary() -> Option<usize> { Some(0) }\n", encoding="utf-8")
+        guards = REPOSITORY_POLICY["categories"]["replay"]["guards"]
+
+        def matches(guard: dict[str, object]) -> bool:
+            return VALIDATOR._guard_matches(self.root, "HEAD", {source}, guard)
+
+        self.assertFalse(any(matches(guard) for guard in guards if guard["glob"] != source))
+        self.assertTrue(any(matches(guard) for guard in guards))
+        self.assertFalse(any(
+            VALIDATOR._guard_matches(self.root, "HEAD", {"src/replay_clock/tests/parallel.rs"}, guard)
+            for guard in guards
+        ))
+
     def test_history_fixture_in_unchanged_test_module_is_not_production_replay(self) -> None:
         diff = """\
 diff --git a/src/lib.rs b/src/lib.rs
diff --git a/src/replay_clock.rs b/src/replay_clock.rs
index 1c2f343..29d6c5b 100644
--- a/src/replay_clock.rs
+++ b/src/replay_clock.rs
@@ -14,6 +14,7 @@ use std::{
 
 mod condition;
 mod condition_timeout;
+mod parallel;
 mod parallel_activity;
 mod parallel_timer;
 
diff --git a/src/replay_clock/parallel.rs b/src/replay_clock/parallel.rs
new file mode 100644
index 0000000..1ea49f3
--- /dev/null
+++ b/src/replay_clock/parallel.rs
@@ -0,0 +1,59 @@
+//! Portable v1 group time. Legacy Carbon observations remain separate evidence.
+use super::{durable_event_sequence, parse, DateTime, HashMap, HistoryEvent, Utc, Value};
+use crate::ParallelGroupMetadata;
+
+pub(super) fn all_path(path: &[ParallelGroupMetadata], kind: &str) -> bool {
+    !path.is_empty()
+        && path.iter().all(|entry| {
+            entry.parallel_group_kind == kind
+                && entry
+                    .parallel_group_mode
+                    .as_deref()
+                    .is_none_or(|mode| mode == "all")
+                && entry.selection_member_key.is_none()
+                && entry.selection_member_index.is_none()
+                && entry.selection_member_base_sequence.is_none()
+                && entry.selection_member_size.is_none()
+                && entry.selection_member_kind.is_none()
+        })
+}
+
+pub(super) fn round_microseconds(micros: i64) -> Option<i64> {
+    let magnitude = micros.unsigned_abs().checked_add(500)? / 1000;
+    let milliseconds = i64::try_from(magnitude).ok()?;
+    if micros < 0 {
+        milliseconds.checked_neg()
+    } else {
+        Some(milliseconds)
+    }
+}
+
+pub(super) fn milliseconds(time: DateTime<Utc>) -> Option<i64> {
+    if time.timestamp_subsec_nanos() % 1000 != 0 {
+        return None;
+    }
+    let micros = time
+        .timestamp()
+        .checked_mul(1_000_000)?
+        .checked_add(i64::from(time.timestamp_subsec_micros()))?;
+    round_microseconds(micros)
+}
+
+pub(super) fn event_time(event: &HistoryEvent) -> Option<DateTime<Utc>> {
+    match event.raw.get("recorded_at") {
+        None | Some(Value::Null) => parse(event.raw.get("created_at")),
+        present => parse(present),
+    }
+}
+
+pub(super) fn event_indices(history: &[HistoryEvent], prefix: &str) -> HashMap<u64, Vec<usize>> {
+    let mut events: HashMap<u64, Vec<usize>> = HashMap::new();
+    for (index, event) in history.iter().enumerate() {
+        if event.event_type.starts_with(prefix) {
+            if let Some(sequence) = durable_event_sequence(event) {
+                events.entry(sequence).or_default().push(index);
+            }
+        }
+    }
+    events
+}
diff --git a/src/replay_clock/parallel_activity.rs b/src/replay_clock/parallel_activity.rs
index b841825..07568a9 100644
--- a/src/replay_clock/parallel_activity.rs
+++ b/src/replay_clock/parallel_activity.rs
@@ -1,6 +1,5 @@
-use super::{
-    durable_event_sequence, parse, DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value,
-};
+use super::parallel::{all_path, event_indices, event_time, milliseconds};
+use super::{DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value};
 use crate::{recorded_parallel_group_path, ParallelGroupMetadata};
 use std::collections::HashSet;
 
@@ -15,22 +14,6 @@ struct Member {
 #[derive(Debug)]
 pub(super) struct Members(HashMap<u64, Member>);
 
-pub(super) fn all_path(path: &[ParallelGroupMetadata], kind: &str) -> bool {
-    !path.is_empty()
-        && path.iter().all(|entry| {
-            entry.parallel_group_kind == kind
-                && entry
-                    .parallel_group_mode
-                    .as_deref()
-                    .is_none_or(|mode| mode == "all")
-                && entry.selection_member_key.is_none()
-                && entry.selection_member_index.is_none()
-                && entry.selection_member_base_sequence.is_none()
-                && entry.selection_member_size.is_none()
-                && entry.selection_member_kind.is_none()
-        })
-}
-
 fn member(
     history: &[HistoryEvent],
     indices: &[usize],
@@ -99,34 +82,9 @@ fn member(
     })
 }
 
-// Carbon 3.15.1 Timestamp::getTimestampMs calls round(microseconds / 1000),
-// not truncation. Only accept the range where its float microsecond input is
-// exact, and the persisted microsecond precision the Core carrier represents.
-pub(super) fn milliseconds(time: DateTime<Utc>) -> Option<i64> {
-    let micros = time.timestamp_micros();
-    if micros.unsigned_abs() > (1_u64 << 53) || time.timestamp_subsec_nanos() % 1000 != 0 {
-        return None;
-    }
-    Some((micros as f64 / 1000.0).round() as i64)
-}
-
-pub(super) fn event_time(event: &HistoryEvent) -> Option<DateTime<Utc>> {
-    match event.raw.get("recorded_at") {
-        None | Some(Value::Null) => parse(event.raw.get("created_at")),
-        present => parse(present),
-    }
-}
-
 impl Members {
     pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self {
-        let mut events: HashMap<u64, Vec<usize>> = HashMap::new();
-        for (index, event) in history.iter().enumerate() {
-            if event.event_type.starts_with("Activity") {
-                if let Some(sequence) = durable_event_sequence(event) {
-                    events.entry(sequence).or_default().push(index);
-                }
-            }
-        }
+        let events = event_indices(history, "Activity");
         Self(
             commands
                 .iter()
@@ -193,10 +151,10 @@ impl Members {
         }
         match failed {
             Some(selected) => {
-                let (ms, core_selected) = failure?;
-                // Keep the legacy exception even when Core selects a different
+                let (ms, canonical_selected) = failure?;
+                // Keep the legacy exception even when v1 selects a different
                 // failed member. That compatibility gap has no clock authority.
-                (selected == core_selected)
+                (selected == canonical_selected)
                     .then(|| DateTime::from_timestamp_millis(ms))
                     .flatten()
             }
diff --git a/src/replay_clock/parallel_timer.rs b/src/replay_clock/parallel_timer.rs
index e696816..81bc4a8 100644
--- a/src/replay_clock/parallel_timer.rs
+++ b/src/replay_clock/parallel_timer.rs
@@ -1,5 +1,5 @@
-use super::parallel_activity::{all_path, event_time, milliseconds};
-use super::{durable_event_sequence, DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value};
+use super::parallel::{all_path, event_indices, event_time, milliseconds};
+use super::{DateTime, HashMap, HistoryEvent, RecordedCommand, Utc, Value};
 use crate::{recorded_parallel_group_path, ParallelGroupMetadata};
 use std::collections::HashSet;
 
@@ -65,14 +65,7 @@ fn member(
 
 impl Members {
     pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self {
-        let mut events: HashMap<u64, Vec<usize>> = HashMap::new();
-        for (index, event) in history.iter().enumerate() {
-            if event.event_type.starts_with("Timer") {
-                if let Some(sequence) = durable_event_sequence(event) {
-                    events.entry(sequence).or_default().push(index);
-                }
-            }
-        }
+        let events = event_indices(history, "Timer");
         Self(
             commands
                 .iter()
diff --git a/src/replay_clock/tests.rs b/src/replay_clock/tests.rs
index 0ac9d1d..a60a6c8 100644
--- a/src/replay_clock/tests.rs
+++ b/src/replay_clock/tests.rs
@@ -524,3 +524,5 @@ mod parallel_activity;
 mod condition_timeout;
 
 mod parallel_timer;
+
+mod parallel;
diff --git a/src/replay_clock/tests/parallel.rs b/src/replay_clock/tests/parallel.rs
new file mode 100644
index 0000000..b1ae1cb
--- /dev/null
+++ b/src/replay_clock/tests/parallel.rs
@@ -0,0 +1,52 @@
+use super::*;
+
+#[test]
+fn replay_clock_canonical_v1_uses_exact_signed_microseconds_across_epoch_boundaries_and_offsets() {
+    let table: Value = serde_json::from_str(include_str!(
+        "../../../tests/fixtures/replay-clock-canonical-v1.json"
+    ))
+    .unwrap();
+    for row in table["vectors"].as_array().unwrap() {
+        let time = DateTime::parse_from_rfc3339(row["input"].as_str().unwrap())
+            .unwrap()
+            .with_timezone(&Utc);
+        assert_eq!(
+            time.timestamp_micros().to_string(),
+            row["utc_microseconds"].as_str().unwrap()
+        );
+        let ms = super::super::parallel::milliseconds(time).unwrap();
+        assert_eq!(
+            ms.to_string(),
+            row["milliseconds"].as_str().unwrap(),
+            "{}",
+            row["id"]
+        );
+        assert_eq!(
+            DateTime::<Utc>::from_timestamp_millis(ms)
+                .unwrap()
+                .to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
+            row["reconstructed"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn replay_clock_canonical_v1_handles_signed_integer_limits_without_float_or_abs_overflow() {
+    use super::super::parallel::round_microseconds;
+    for (micro, ms) in [
+        (-501, -1),
+        (-500, -1),
+        (-499, 0),
+        (499, 0),
+        (500, 1),
+        (501, 1),
+        (i64::MAX, 9_223_372_036_854_776),
+        (i64::MIN, -9_223_372_036_854_776),
+    ] {
+        assert_eq!(round_microseconds(micro), Some(ms));
+    }
+    let too_precise = DateTime::parse_from_rfc3339("2026-01-01T00:00:00.123456789Z")
+        .unwrap()
+        .with_timezone(&Utc);
+    assert!(super::super::parallel::milliseconds(too_precise).is_none());
+}
diff --git a/src/replay_clock/tests/parallel_activity.rs b/src/replay_clock/tests/parallel_activity.rs
index 4e97494..1a1f3cb 100644
--- a/src/replay_clock/tests/parallel_activity.rs
+++ b/src/replay_clock/tests/parallel_activity.rs
@@ -214,7 +214,7 @@ fn replay_now_group_requires_all_success_times_and_never_substitutes_other_autho
         observe(history, false)["time"],
         json!("2026-01-01T00:00:10.123456Z")
     );
-    for time in ["2500-01-01T00:00:00Z", "2026-01-01T00:00:00.123456789Z"] {
+    for time in ["2026-01-01T00:00:00.123456789Z"] {
         assert!(
             observe(group(1, ["ActivityCompleted"; 2], [time; 2], false), false)["time"].is_null()
         );
@@ -299,6 +299,8 @@ async fn replay_now_group_query_does_not_borrow_executor_aggregate() {
 
 #[test]
 fn replay_now_group_carbon_reference_matches_measured_rounding_and_reconstruction() {
+    // These attributed PHP 8.4/Carbon observations coincide with v1 on these
+    // inputs. They are not universal runtime or producer-context authority.
     let reference: Value = serde_json::from_str(include_str!(
         "../../../tests/fixtures/replay-clock-carbon-reference.json"
     ))
@@ -307,7 +309,7 @@ fn replay_now_group_carbon_reference_matches_measured_rounding_and_reconstructio
         let time = DateTime::parse_from_rfc3339(row["input"].as_str().unwrap())
             .unwrap()
             .with_timezone(&Utc);
-        let ms = super::super::parallel_activity::milliseconds(time).unwrap();
+        let ms = super::super::parallel::milliseconds(time).unwrap();
         assert_eq!(json!(ms), row["milliseconds"]);
         assert_eq!(
             DateTime::<Utc>::from_timestamp_millis(ms)
diff --git a/src/replay_clock/tests/parallel_timer.rs b/src/replay_clock/tests/parallel_timer.rs
index 9e9a50a..2ce3724 100644
--- a/src/replay_clock/tests/parallel_timer.rs
+++ b/src/replay_clock/tests/parallel_timer.rs
@@ -123,7 +123,7 @@ fn replay_now_timer_all_missing_or_invalid_time_has_no_scalar_fallback() {
         observe(history, false),
         json!("2026-01-01T00:00:10.654321Z")
     );
-    for time in ["2500-01-01T00:00:00Z", "2026-01-01T00:00:00.123456789Z"] {
+    for time in ["2026-01-01T00:00:00.123456789Z"] {
         assert!(observe(group(1, [time; 2], false), false).is_null());
     }
 }
diff --git a/tests/fixtures/replay-clock-canonical-v1.json b/tests/fixtures/replay-clock-canonical-v1.json
new file mode 100644
index 0000000..1aae5d7
--- /dev/null
+++ b/tests/fixtures/replay-clock-canonical-v1.json
@@ -0,0 +1,189 @@
+{
+  "schema": "durable-workflow.replay-clock-quantization/v1",
+  "arithmetic": "sign(U) * ((abs(U) + 500) div 1000)",
+  "integer_fields": "decimal strings preserve exact values across JSON runtimes",
+  "vectors": [
+    {
+      "id": "canonical-v1-01",
+      "input": "1970-01-01T00:00:00.000000Z",
+      "utc_microseconds": "0",
+      "milliseconds": "0",
+      "reconstructed": "1970-01-01T00:00:00.000000Z"
+    },
+    {
+      "id": "canonical-v1-02",
+      "input": "1970-01-01T00:00:00.000499Z",
+      "utc_microseconds": "499",
+      "milliseconds": "0",
+      "reconstructed": "1970-01-01T00:00:00.000000Z"
+    },
+    {
+      "id": "canonical-v1-03",
+      "input": "1970-01-01T00:00:00.000500Z",
+      "utc_microseconds": "500",
+      "milliseconds": "1",
+      "reconstructed": "1970-01-01T00:00:00.001000Z"
+    },
+    {
+      "id": "canonical-v1-04",
+      "input": "1970-01-01T00:00:00.000501Z",
+      "utc_microseconds": "501",
+      "milliseconds": "1",
+      "reconstructed": "1970-01-01T00:00:00.001000Z"
+    },
+    {
+      "id": "canonical-v1-05",
+      "input": "1969-12-31T23:59:59.999501Z",
+      "utc_microseconds": "-499",
+      "milliseconds": "0",
+      "reconstructed": "1970-01-01T00:00:00.000000Z"
+    },
+    {
+      "id": "canonical-v1-06",
+      "input": "1969-12-31T23:59:59.999500Z",
+      "utc_microseconds": "-500",
+      "milliseconds": "-1",
+      "reconstructed": "1969-12-31T23:59:59.999000Z"
+    },
+    {
+      "id": "canonical-v1-07",
+      "input": "1969-12-31T23:59:59.999499Z",
+      "utc_microseconds": "-501",
+      "milliseconds": "-1",
+      "reconstructed": "1969-12-31T23:59:59.999000Z"
+    },
+    {
+      "id": "canonical-v1-08",
+      "input": "2026-01-01T00:00:00.123400Z",
+      "utc_microseconds": "1767225600123400",
+      "milliseconds": "1767225600123",
+      "reconstructed": "2026-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-09",
+      "input": "2026-01-01T00:00:00.123499Z",
+      "utc_microseconds": "1767225600123499",
+      "milliseconds": "1767225600123",
+      "reconstructed": "2026-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-10",
+      "input": "2026-01-01T00:00:00.123500Z",
+      "utc_microseconds": "1767225600123500",
+      "milliseconds": "1767225600124",
+      "reconstructed": "2026-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-11",
+      "input": "2026-01-01T00:00:00.123501Z",
+      "utc_microseconds": "1767225600123501",
+      "milliseconds": "1767225600124",
+      "reconstructed": "2026-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-12",
+      "input": "2026-01-01T01:30:00.123500+01:30",
+      "utc_microseconds": "1767225600123500",
+      "milliseconds": "1767225600124",
+      "reconstructed": "2026-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-13",
+      "input": "2026-01-01T00:00:00.123500-00:00",
+      "utc_microseconds": "1767225600123500",
+      "milliseconds": "1767225600124",
+      "reconstructed": "2026-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-14",
+      "input": "1900-01-01T00:00:00.123499Z",
+      "utc_microseconds": "-2208988799876501",
+      "milliseconds": "-2208988799877",
+      "reconstructed": "1900-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-15",
+      "input": "1900-01-01T00:00:00.123500Z",
+      "utc_microseconds": "-2208988799876500",
+      "milliseconds": "-2208988799877",
+      "reconstructed": "1900-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-16",
+      "input": "1900-01-01T00:00:00.123501Z",
+      "utc_microseconds": "-2208988799876499",
+      "milliseconds": "-2208988799876",
+      "reconstructed": "1900-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-17",
+      "input": "1000-01-01T00:00:00.123499Z",
+      "utc_microseconds": "-30610223999876501",
+      "milliseconds": "-30610223999877",
+      "reconstructed": "1000-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-18",
+      "input": "1000-01-01T00:00:00.123500Z",
+      "utc_microseconds": "-30610223999876500",
+      "milliseconds": "-30610223999877",
+      "reconstructed": "1000-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-19",
+      "input": "1000-01-01T00:00:00.123501Z",
+      "utc_microseconds": "-30610223999876499",
+      "milliseconds": "-30610223999876",
+      "reconstructed": "1000-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-20",
+      "input": "3000-01-01T00:00:00.123499Z",
+      "utc_microseconds": "32503680000123499",
+      "milliseconds": "32503680000123",
+      "reconstructed": "3000-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-21",
+      "input": "3000-01-01T00:00:00.123500Z",
+      "utc_microseconds": "32503680000123500",
+      "milliseconds": "32503680000124",
+      "reconstructed": "3000-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-22",
+      "input": "3000-01-01T00:00:00.123501Z",
+      "utc_microseconds": "32503680000123501",
+      "milliseconds": "32503680000124",
+      "reconstructed": "3000-01-01T00:00:00.124000Z"
+    },
+    {
+      "id": "canonical-v1-23",
+      "input": "0001-01-01T00:00:00.000499Z",
+      "utc_microseconds": "-62135596799999501",
+      "milliseconds": "-62135596800000",
+      "reconstructed": "0001-01-01T00:00:00.000000Z"
+    },
+    {
+      "id": "canonical-v1-24",
+      "input": "0001-01-01T00:00:00.000500Z",
+      "utc_microseconds": "-62135596799999500",
+      "milliseconds": "-62135596800000",
+      "reconstructed": "0001-01-01T00:00:00.000000Z"
+    },
+    {
+      "id": "canonical-v1-25",
+      "input": "9999-01-01T00:00:00.123499Z",
+      "utc_microseconds": "253370764800123499",
+      "milliseconds": "253370764800123",
+      "reconstructed": "9999-01-01T00:00:00.123000Z"
+    },
+    {
+      "id": "canonical-v1-26",
+      "input": "9999-01-01T00:00:00.123500Z",
+      "utc_microseconds": "253370764800123500",
+      "milliseconds": "253370764800124",
+      "reconstructed": "9999-01-01T00:00:00.124000Z"
+    }
+  ]
+}
diff --git a/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json b/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json
new file mode 100644
index 0000000..01fbf87
--- /dev/null
+++ b/tests/fixtures/replay-regressions/business-replay-clock-canonical-v1-large-epochs.json
@@ -0,0 +1,251 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "rust-business-replay-clock-canonical-v1-large-epochs",
+  "protocol_version": "1.20",
+  "bindings": [
+    "rust"
+  ],
+  "workflow": {
+    "type": "corpus.business-replay-clock-parallel-activity",
+    "input": [],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 1,
+        "activity_type": "clock-a",
+        "activity_execution_id": "clock-execution-1",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 1
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 2,
+        "activity_type": "clock-b",
+        "activity_execution_id": "clock-execution-2",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 2
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "sequence": 1,
+        "activity_type": "clock-a",
+        "activity_execution_id": "clock-execution-1",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ],
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoGb25l"
+        }
+      },
+      "recorded_at": "3000-01-01T00:00:05.123499Z",
+      "sequence": 3
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "sequence": 2,
+        "activity_type": "clock-b",
+        "activity_execution_id": "clock-execution-2",
+        "parallel_group_id": "parallel-activities:1:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 1,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:1:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 1,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ],
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoGb25l"
+        }
+      },
+      "recorded_at": "3000-01-01T00:00:05.123500Z",
+      "sequence": 4
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 3,
+        "activity_type": "clock-a",
+        "activity_execution_id": "clock-execution-3",
+        "parallel_group_id": "parallel-activities:3:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 3,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:3:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 3,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 5
+    },
+    {
+      "event_type": "ActivityScheduled",
+      "payload": {
+        "sequence": 4,
+        "activity_type": "clock-b",
+        "activity_execution_id": "clock-execution-4",
+        "parallel_group_id": "parallel-activities:3:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 3,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:3:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 3,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ]
+      },
+      "recorded_at": "2026-01-01T00:00:00Z",
+      "sequence": 6
+    },
+    {
+      "event_type": "ActivityFailed",
+      "payload": {
+        "sequence": 3,
+        "activity_type": "clock-a",
+        "activity_execution_id": "clock-execution-3",
+        "parallel_group_id": "parallel-activities:3:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 3,
+        "parallel_group_size": 2,
+        "parallel_group_index": 0,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:3:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 3,
+            "parallel_group_size": 2,
+            "parallel_group_index": 0
+          }
+        ],
+        "message": "recorded group failure"
+      },
+      "recorded_at": "1000-01-01T00:00:03.123500Z",
+      "sequence": 7
+    },
+    {
+      "event_type": "ActivityCompleted",
+      "payload": {
+        "sequence": 4,
+        "activity_type": "clock-b",
+        "activity_execution_id": "clock-execution-4",
+        "parallel_group_id": "parallel-activities:3:2",
+        "parallel_group_kind": "activity",
+        "parallel_group_base_sequence": 3,
+        "parallel_group_size": 2,
+        "parallel_group_index": 1,
+        "parallel_group_path": [
+          {
+            "parallel_group_id": "parallel-activities:3:2",
+            "parallel_group_kind": "activity",
+            "parallel_group_base_sequence": 3,
+            "parallel_group_size": 2,
+            "parallel_group_index": 1
+          }
+        ],
+        "result": {
+          "codec": "avro",
+          "blob": "wwHioz3/VYAiNwoGb25l"
+        }
+      },
+      "recorded_at": "3000-01-01T00:00:00Z",
+      "sequence": 8
+    }
+  ],
+  "expected": {
+    "type": "complete_workflow",
+    "result": {
+      "entry_unavailable": true,
+      "success_time": "3000-01-01T00:00:05.123500Z",
+      "failure_time": "1000-01-01T00:00:03.123000Z",
+      "failure_member": [
+        0
+      ],
+      "unchanged_prefix": true
+    }
+  },
+  "command_sequence": [
+    {
+      "type": "complete_workflow",
+      "result": {
+        "entry_unavailable": true,
+        "success_time": "3000-01-01T00:00:05.123500Z",
+        "failure_time": "1000-01-01T00:00:03.123000Z",
+        "failure_member": [
+          0
+        ],
+        "unchanged_prefix": true
+      }
+    }
+  ],
+  "worker_task": {
+    "workflow_id": "parallel-activity-clock-parent",
+    "payload_codec": "avro"
+  }
+}