fix: expose consumed parallel activity replay time

docs/replay-clock-source.md

@@ -24,12 +24,13 @@ 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 |
 | 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 supported boundary rules |
+| 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 |
 | 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 timeout result, grouped/selection/Nexus or cancellation callback | Their dedicated clock algorithms are outside this initial cohort | Explicitly unavailable; existing execution/cancellation behavior remains intact |
+| Signal timeout result, other 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
@@ -131,3 +132,37 @@ time invalidates the consumed boundary instead of borrowing the previous one.
 Grouped conditions and their dedicated prefix algorithms remain outside this
 scalar cohort. Canonical Rust occurrence fields are existing parser authority;
 this source adds no occurrence field or producer representation to Core.
+
+Persisted parallel Activity `all` follows [Core's descriptor aggregation](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L1916)
+and [the accepted cross-SDK decision](https://github.com/durable-workflow/.github/issues/215#issuecomment-6677422213).
+Only consumed members of that authored group participate. Each needs a unique
+ordered schedule/terminal lifecycle, explicit numeric counter, exact type,
+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
+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.
+Scalar redrive reuse timestamps, a later stream row, unconsumed groups and an
+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
+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.
+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.

regression-corpus-policy.json

@@ -74,6 +74,9 @@
         },
         {
           "glob": "src/replay_clock/condition.rs"
+        },
+        {
+          "glob": "src/replay_clock/parallel_activity.rs"
         }
       ]
     }

scripts/ci/test-validate-regression-corpus.py

@@ -663,6 +663,23 @@ def matches(guard: dict[str, object]) -> bool:
             for guard in guards
         ))
 
+    def test_parallel_activity_clock_module_requires_replay_evidence(self) -> None:
+        source = "src/replay_clock/parallel_activity.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_activity.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

src/lib.rs

@@ -9682,7 +9682,7 @@ impl Worker {
                 };
                 query_authority::before_handler(authority, &self.client).await?;
                 let mut invocation = {
-                    let _clock = replay_clock::Guard::enter(&workflow_context);
+                    let _clock = replay_clock::Guard::query(&workflow_context);
                     replay(workflow_context.clone(), workflow_input_typed.clone())
                 }
                 .map_err(|error| {
@@ -9694,7 +9694,7 @@ impl Worker {
                 })?;
                 let outcome = {
                     let mut cx = TaskContext::from_waker(noop_waker_ref());
-                    let _clock = replay_clock::Guard::enter(&workflow_context);
+                    let _clock = replay_clock::Guard::query(&workflow_context);
                     let _replay = cancellation_replay_clock::ReplayGuard::enter(&workflow_context)
                         .map_err(|error| {
                             QueryTaskExecutionFailure::new(

src/replay_clock.rs

@@ -7,32 +7,46 @@ use super::{
 };
 use chrono::{DateTime, Utc};
 use std::{
-    cell::RefCell,
+    cell::{Cell, RefCell},
     collections::HashMap,
     sync::{Arc, Mutex, Weak},
 };
 
 mod condition;
+mod parallel_activity;
 
 thread_local! {
     static ACTIVE: RefCell<Option<Weak<Mutex<WorkflowState>>>> = const { RefCell::new(None) };
+    static QUERY_REPLAY: Cell<bool> = const { Cell::new(false) };
 }
 
-pub(super) struct Guard(Option<Weak<Mutex<WorkflowState>>>);
+pub(super) struct Guard(Option<Weak<Mutex<WorkflowState>>>, bool);
 
 impl Guard {
     pub(super) fn enter(context: &WorkflowContext) -> Self {
-        Self(ACTIVE.with(|active| active.replace(Some(Arc::downgrade(&context.state)))))
+        Self::scope(Some(Arc::downgrade(&context.state)), false)
+    }
+
+    pub(super) fn query(context: &WorkflowContext) -> Self {
+        Self::scope(Some(Arc::downgrade(&context.state)), true)
     }
 
     pub(super) fn unavailable() -> Self {
-        Self(ACTIVE.with(|active| active.replace(None)))
+        Self::scope(None, false)
+    }
+
+    fn scope(context: Option<Weak<Mutex<WorkflowState>>>, query: bool) -> Self {
+        Self(
+            ACTIVE.with(|active| active.replace(context)),
+            QUERY_REPLAY.with(|active| active.replace(query)),
+        )
     }
 }
 
 impl Drop for Guard {
     fn drop(&mut self) {
         ACTIVE.with(|active| active.replace(self.0.take()));
+        QUERY_REPLAY.with(|active| active.set(self.1));
     }
 }
 
@@ -76,6 +90,7 @@ fn start_time(history: &[HistoryEvent]) -> Option<DateTime<Utc>> {
 pub(super) struct Clock {
     time: Option<DateTime<Utc>>,
     boundaries: HashMap<u64, usize>,
+    parallel_activities: parallel_activity::Members,
 }
 
 enum BoundaryKind {
@@ -365,6 +380,7 @@ impl Clock {
         Self {
             time: start_time(history),
             boundaries,
+            parallel_activities: parallel_activity::Members::new(history, commands),
         }
     }
 
@@ -393,7 +409,17 @@ impl WorkflowState {
     }
 
     pub(super) fn consume_replay_group(&mut self, sequences: &[u64], failed: Option<u64>) {
-        self.replay_clock.unavailable();
+        // Core's query group-send branch supplies no aggregate time. Do not
+        // transfer executor authority into reconstructed query execution.
+        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.advance_cancellation_sequences(sequences, failed);
     }
 
@@ -448,8 +474,8 @@ impl WorkflowContext {
     /// is available only while this workflow's handler/future is executing.
     ///
     /// 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
+    /// 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.
     /// Migrate business decisions through normal workflow versioning. Event
     /// timestamps do not provide a total order; retain stream/counter identity.