fix: correlate fired condition timeout replay time

docs/replay-clock-source.md

@@ -25,6 +25,7 @@ counter identity to order emitted events.
 | 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 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 |
 | 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 |
@@ -122,8 +123,7 @@ numeric counters and complete matching definitions; legacy fallback parsing
 may still return a result when that time authority is missing. Each actual
 physical resolution in a logical reopen chain advances only when consumed.
 
-A timeout fire without its persisted condition resolution supplies no new
-condition clock. Core's live timeout branch has its own timer/fired-row time
+The explicit resolution branch does not borrow a timeout fire's time. Core's live timeout branch has its own timer/fired-row time
 input; it is not inferred for Rust from an unrelated timer. Predicate callbacks
 still have explicit unavailability inside the invocation. Synchronous local
 true/zero-timeout paths preserve the already accepted previous-prefix behavior,
@@ -166,3 +166,30 @@ 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.
+
+The persisted fired-only condition-timeout cohort follows Core
+[`resumeTimedOutCondition`](https://github.com/durable-workflow/workflow/blob/ce24207ef86856e64f97b939c9aeef9d1fb387c8/src/V2/Support/WorkflowExecutor.php#L895).
+It consumes the selected `TimerFired.recorded_at`, including its precision.
+The native producer already carries the matching physical/logical wait,
+key/fingerprint, timer ID/kind, command sequence and delay through
+[`wait_fields`](https://github.com/durable-workflow/server/blob/fadf8729265e077f280c8f71a87c06d8f73a682e/rust/src/runtime/wait_timeouts.rs#L38)
+and the scheduled/fired history. It intentionally need not append a
+`ConditionWaitTimedOut` event. This source change adds no field or producer.
+
+One ordered open/schedule/fire and complete matching metadata are required.
+A cancelled, aliased, duplicate, reordered, superseded or otherwise ambiguous
+physical wait cannot provide time. A logical reopen uses the final actually
+consumed physical wait; an old orphan timeout cannot supply a pending reopen's
+clock. Indexing reads identities once, and accessors only read the consumed
+boundary. Payload `fired_at`, deadline, reuse, insertion aliases and an
+unrepresented timer-row fallback do not replace the selected `recorded_at`.
+Missing or invalid time is unavailable while legacy results remain unchanged.
+
+This fired-only authority is executor-specific. Current Core query replay stops
+without an explicit condition resolution, so snapshot and reconstructed-query
+clocks remain unavailable on this shape, including inside the reconstructed
+workflow. Existing explicit persisted condition-resolution query behavior is
+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.

regression-corpus-policy.json

@@ -77,6 +77,9 @@
         },
         {
           "glob": "src/replay_clock/parallel_activity.rs"
+        },
+        {
+          "glob": "src/replay_clock/condition_timeout.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_condition_timeout_clock_module_requires_replay_evidence(self) -> None:
+        source = "src/replay_clock/condition_timeout.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/condition_timeout.rs"}, guard)
+            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

src/replay_clock.rs

@@ -13,6 +13,7 @@ use std::{
 };
 
 mod condition;
+mod condition_timeout;
 mod parallel_activity;
 
 thread_local! {
@@ -73,6 +74,26 @@ fn recorded_time(event: &HistoryEvent, reused: bool) -> Option<DateTime<Utc>> {
     )
 }
 
+fn boundary_time(event: &HistoryEvent, query: bool) -> Option<DateTime<Utc>> {
+    let internal = event.event_type == "TimerFired" && is_internal_timer_event(event);
+    if internal {
+        // Core query replay requires the explicit condition resolution. Its
+        // executor-only fired-time branch does not grant callback authority.
+        return if query {
+            None
+        } else {
+            parse(event.raw.get("recorded_at"))
+        };
+    }
+    recorded_time(
+        event,
+        matches!(
+            event.event_type.as_str(),
+            "ActivityCompleted" | "TimerFired" | "SignalApplied"
+        ),
+    )
+}
+
 fn start_time(history: &[HistoryEvent]) -> Option<DateTime<Utc>> {
     let mut starts = history
         .iter()
@@ -372,8 +393,11 @@ impl Clock {
                 boundaries.insert(sequence, index);
             }
         }
+        let condition_timeouts = condition_timeout::Index::new(history);
         boundaries.extend(conditions.into_iter().filter_map(|(sequence, indices)| {
-            condition::boundary(history, &indices, sequence).map(|index| (sequence, index))
+            condition::boundary(history, &indices, sequence)
+                .or_else(|| condition_timeouts.boundary(history, &indices, sequence))
+                .map(|index| (sequence, index))
         }));
         // Index identities once. Timestamps are read only when that boundary
         // is actually consumed, never by scanning ahead from the accessor.
@@ -453,15 +477,7 @@ impl WorkflowState {
             .boundaries
             .get(&sequence)
             .and_then(|index| self.history_events.get(*index))
-            .and_then(|event| {
-                recorded_time(
-                    event,
-                    matches!(
-                        event.event_type.as_str(),
-                        "ActivityCompleted" | "TimerFired" | "SignalApplied"
-                    ),
-                )
-            });
+            .and_then(|event| boundary_time(event, QUERY_REPLAY.with(Cell::get)));
     }
 }
 
@@ -536,15 +552,7 @@ pub(super) fn snapshot_time(history: &[HistoryEvent], codec: &str) -> Option<Dat
         .boundaries
         .get(&durable_event_sequence(event)?)
         .and_then(|index| history.get(*index))
-        .and_then(|event| {
-            recorded_time(
-                event,
-                matches!(
-                    event.event_type.as_str(),
-                    "ActivityCompleted" | "TimerFired" | "SignalApplied"
-                ),
-            )
-        })
+        .and_then(|event| boundary_time(event, true))
 }
 
 pub(super) fn update_time(history: &[HistoryEvent], update_id: &str) -> Option<DateTime<Utc>> {

src/replay_clock/condition_timeout.rs

@@ -0,0 +1,146 @@
+use super::{HashMap, HistoryEvent, Value};
+use crate::recorded_parallel_group_path;
+
+fn counter(event: &HistoryEvent, sequence: u64) -> bool {
+    event.payload.get("sequence").and_then(Value::as_u64) == Some(sequence)
+        && event
+            .payload
+            .get("workflow_sequence")
+            .is_none_or(|value| value.as_u64() == Some(sequence))
+}
+
+pub(super) struct Index<'a> {
+    timers_by_wait: HashMap<&'a str, Vec<usize>>,
+    timers_by_id: HashMap<&'a str, Vec<usize>>,
+    conditions_by_wait: HashMap<&'a str, Vec<usize>>,
+    opens_by_occurrence: HashMap<&'a str, Vec<usize>>,
+}
+
+fn link<'a>(
+    map: &mut HashMap<&'a str, Vec<usize>>,
+    event: &'a HistoryEvent,
+    key: &str,
+    position: usize,
+) {
+    if let Some(value) = event.payload.get(key).and_then(Value::as_str) {
+        map.entry(value).or_default().push(position);
+    }
+}
+
+impl<'a> Index<'a> {
+    pub(super) fn new(history: &'a [HistoryEvent]) -> Self {
+        let mut index = Self {
+            timers_by_wait: HashMap::new(),
+            timers_by_id: HashMap::new(),
+            conditions_by_wait: HashMap::new(),
+            opens_by_occurrence: HashMap::new(),
+        };
+        for (position, event) in history.iter().enumerate() {
+            if event.event_type.starts_with("Timer") {
+                link(
+                    &mut index.timers_by_wait,
+                    event,
+                    "condition_wait_id",
+                    position,
+                );
+                link(&mut index.timers_by_id, event, "timer_id", position);
+            }
+            if event.event_type.starts_with("ConditionWait") {
+                link(
+                    &mut index.conditions_by_wait,
+                    event,
+                    "condition_wait_id",
+                    position,
+                );
+            }
+            if event.event_type == "ConditionWaitOpened" {
+                link(
+                    &mut index.opens_by_occurrence,
+                    event,
+                    "condition_wait_occurrence_id",
+                    position,
+                );
+            }
+        }
+        index
+    }
+
+    pub(super) fn boundary(
+        &self,
+        history: &[HistoryEvent],
+        indices: &[usize],
+        sequence: u64,
+    ) -> Option<usize> {
+        // A persisted explicit resolution has its own, different Core branch.
+        let [opened] = indices else { return None };
+        let open = &history[*opened];
+        if open.event_type != "ConditionWaitOpened" || !counter(open, sequence) {
+            return None;
+        }
+        let identity = [
+            "condition_wait_id",
+            "condition_wait_occurrence_id",
+            "condition_key",
+            "condition_definition_fingerprint",
+        ];
+        for key in identity {
+            open.payload
+                .get(key)?
+                .as_str()
+                .filter(|value| !value.is_empty())?;
+        }
+        let seconds = open.payload.get("timeout_seconds")?.as_u64()?;
+        let wait = open.payload.get("condition_wait_id")?.as_str()?;
+        if self.conditions_by_wait.get(wait)?.as_slice() != indices {
+            return None;
+        }
+        let timers = self.timers_by_wait.get(wait)?;
+        let [scheduled, fired] = timers.as_slice() else {
+            return None;
+        };
+        if !(*opened < *scheduled && *scheduled < *fired)
+            || 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 timers {
+            let event = &history[index];
+            if !counter(event, sequence)
+                || event.payload.get("timer_id").and_then(Value::as_str) != Some(timer)
+                || event.payload.get("timer_kind").and_then(Value::as_str)
+                    != Some("condition_timeout")
+                || event.payload.get("delay_seconds").and_then(Value::as_u64) != Some(seconds)
+                || recorded_parallel_group_path(&[event], sequence)
+                    .ok()?
+                    .is_some()
+            {
+                return None;
+            }
+            for key in identity {
+                if event.payload.get(key) != open.payload.get(key) {
+                    return None;
+                }
+            }
+        }
+        // A mislabeled duplicate of this timer must not disappear into another
+        // wait's bucket. This is an identity refusal, never a future-time maximum.
+        if self.timers_by_id.get(timer)? != timers {
+            return None;
+        }
+        let occurrence = open.payload.get("condition_wait_occurrence_id")?.as_str()?;
+        let opens = self.opens_by_occurrence.get(occurrence)?;
+        let next = opens.partition_point(|&index| index <= *opened);
+        if opens.get(next).is_some_and(|&index| index < *fired) {
+            // A later physical open superseded this wait before its old timer
+            // fired. The legacy reopen resolver decides the final call outcome.
+            return None;
+        }
+        Some(*fired)
+    }
+}

src/replay_clock/tests.rs

@@ -520,3 +520,5 @@ mod signal;
 mod condition;
 
 mod parallel_activity;
+
+mod condition_timeout;