Advance replay time for correlated scalar child results

docs/replay-clock-source.md

@@ -1,4 +1,4 @@
-# Additive replay clock: first Rust source cohort
+# Additive replay clock: Rust source cohorts
 
 This source checkpoint implements `WorkflowContext::replay_now()` and
 `QueryContext::replay_now()` toward [the cross-SDK replay-clock feature](https://github.com/durable-workflow/.github/issues/215).
@@ -21,12 +21,13 @@ counter identity to order emitted events.
 | Ordinary entry | Core uses original `WorkflowRun.started_at`; current Rust worker task has no such carrier | Unavailable, even if the start history row has an insertion timestamp |
 | 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 |
-| Snapshot query | Immutable inspected prefix whose final boundary has the same validated activity/timer correlation, or explicit redrive start | Repeated reads of that prefix are equal; unknown/invalid final boundary is unavailable |
+| 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 |
 | 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 |
-| Child/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 |
+| 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 |
 
 The mutable workflow clock is bound only during handler construction and each
 future poll. A separate execution-local guard restores its predecessor on
@@ -57,3 +58,22 @@ also consumed as source evidence, without inventing its ordinary start anchor.
 Normal current-head checks remain required. Published cross-SDK conformance,
 producer implementation/adoption, release and guide/localized delivery remain
 separate work on the owning issue.
+
+The scalar child cohort follows Core's persisted branches at
+[`4b2cdc55` `WorkflowExecutor::resumeChildResolution`](https://github.com/durable-workflow/workflow/blob/4b2cdc5507120d9c213b0b02a85738bec96c574a/src/V2/Support/WorkflowExecutor.php#L1638)
+and the corresponding `QueryStateReplayer` branch. Both completion and failure
+resume at the selected parent resolution event's own `recorded_at`. Unlike the
+successful activity/timer reuse path, they do not select `reused_recorded_at`.
+The child row's `closed_at` is not a substitute. Rust consumes the existing
+validated scalar command before advancing this clock. Identical duplicate
+terminal payloads, missing/conflicting lifecycle identities, changing-run
+retry/continuation lineage, out-of-order lifecycle or invalid/missing time keep
+business time unavailable without changing legacy result/failure behavior.
+A schedule without a resolution stays pending and does not advance time.
+
+Focused child controls exercise actual calls and caught terminal failures,
+unchanged-prefix and cold replay, backward/precise times, future-row isolation,
+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.

regression-corpus-policy.json

@@ -68,6 +68,9 @@
         },
         {
           "glob": "src/cancellation_replay_clock.rs"
+        },
+        {
+          "glob": "src/replay_clock.rs"
         }
       ]
     }

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

@@ -627,6 +627,25 @@ def test_neutral_edit_inside_replay_function_is_related(self) -> None:
 
         self.assertTrue(self._guard_matches(REPLAY_GUARD))
 
+    def test_business_clock_module_requires_replay_evidence(self) -> None:
+        source = "src/replay_clock.rs"
+        (self.root / source).write_text(
+            "fn child_boundary(history: &[usize]) -> Option<usize> { history.last().copied() }\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)
+
+        # Reproduce the original module-coverage gap with the predecessor set.
+        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/child.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/replay_clock.rs

@@ -76,6 +76,54 @@ pub(super) struct Clock {
     boundaries: HashMap<u64, usize>,
 }
 
+enum BoundaryKind {
+    Activity,
+    Timer,
+    Child,
+}
+
+fn child_boundary(history: &[HistoryEvent], indices: &[usize]) -> Option<usize> {
+    // A scalar call has one schedule, an optional start, then one resolution.
+    // Legacy replay can accept duplicate payloads, but their time is ambiguous.
+    let (&scheduled, &terminal) = match indices {
+        [scheduled, terminal] => (scheduled, terminal),
+        [scheduled, started, terminal] if history[*started].event_type == "ChildRunStarted" => {
+            (scheduled, terminal)
+        }
+        _ => return None,
+    };
+    let schedule = &history[scheduled];
+    if schedule.event_type != "ChildWorkflowScheduled"
+        || !matches!(
+            history[terminal].event_type.as_str(),
+            "ChildRunCompleted" | "ChildRunFailed" | "ChildRunCancelled" | "ChildRunTerminated"
+        )
+    {
+        return None;
+    }
+    for key in [
+        "child_call_id",
+        "child_workflow_instance_id",
+        "child_workflow_run_id",
+        "child_workflow_type",
+    ] {
+        let identity = schedule
+            .payload
+            .get(key)?
+            .as_str()
+            .filter(|id| !id.is_empty())?;
+        if indices
+            .iter()
+            .any(|&index| history[index].payload.get(key).and_then(Value::as_str) != Some(identity))
+        {
+            // Changed-run retry/continuation needs its own lineage algorithm.
+            // Missing identities keep legacy result handling, without time.
+            return None;
+        }
+    }
+    Some(terminal)
+}
+
 impl Clock {
     pub(super) fn new(history: &[HistoryEvent], commands: &[RecordedCommand]) -> Self {
         let supported: HashMap<_, _> = commands
@@ -87,39 +135,63 @@ impl Clock {
                     parallel_group_path,
                     ..
                 } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => {
-                    Some((*sequence, true))
+                    Some((*sequence, BoundaryKind::Activity))
                 }
                 RecordedCommand::Timer {
                     sequence,
                     fired: true,
                     parallel_group_path,
                     ..
                 } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => {
-                    Some((*sequence, false))
+                    Some((*sequence, BoundaryKind::Timer))
+                }
+                RecordedCommand::ChildWorkflow {
+                    sequence,
+                    outcome: Some(_),
+                    parallel_group_path,
+                    ..
+                } if parallel_group_path.as_ref().is_none_or(Vec::is_empty) => {
+                    Some((*sequence, BoundaryKind::Child))
                 }
                 _ => None,
             })
             .collect();
         let mut boundaries = HashMap::new();
+        let mut children: HashMap<u64, Vec<usize>> = HashMap::new();
         for (index, event) in history.iter().enumerate() {
             let Some(sequence) = durable_event_sequence(event) else {
                 continue;
             };
             let matched = match supported.get(&sequence) {
-                Some(true) => matches!(
+                Some(BoundaryKind::Activity) => matches!(
                     event.event_type.as_str(),
                     "ActivityCompleted"
                         | "ActivityFailed"
                         | "ActivityTimedOut"
                         | "ActivityCancelled"
                 ),
-                Some(false) => event.event_type == "TimerFired" && !is_internal_timer_event(event),
+                Some(BoundaryKind::Timer) => {
+                    event.event_type == "TimerFired" && !is_internal_timer_event(event)
+                }
+                Some(BoundaryKind::Child) => {
+                    if event.event_type == "ChildWorkflowScheduled"
+                        || event.event_type.starts_with("ChildRun")
+                    {
+                        children.entry(sequence).or_default().push(index);
+                    }
+                    false
+                }
                 None => false,
             };
             if matched {
                 boundaries.entry(sequence).or_insert(index);
             }
         }
+        for (sequence, indices) in children {
+            if let Some(index) = child_boundary(history, &indices) {
+                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 {
@@ -207,7 +279,7 @@ impl WorkflowContext {
     /// or invalid evidence returns [`Error::ReplayClockUnavailable`]. The clock
     /// is available only while this workflow's handler/future is executing.
     ///
-    /// This initial source cohort supports correlated scalar activity/timer
+    /// This source cohort supports correlated scalar activity/timer/child
     /// 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.
@@ -247,6 +319,10 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option<Dat
             | "ActivityTimedOut"
             | "ActivityCancelled"
             | "TimerFired"
+            | "ChildRunCompleted"
+            | "ChildRunFailed"
+            | "ChildRunCancelled"
+            | "ChildRunTerminated"
     ) {
         return None;
     }

src/replay_clock/tests.rs

@@ -513,3 +513,5 @@ async fn replay_now_reconstructed_query_uses_consumed_prefix_not_snapshot_tail()
         json!("2026-01-01T00:00:03.765432Z")
     );
 }
+
+mod child;