diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
index bf7a6056831..6a65cef57e1 100644
--- a/crates/adapters/src/controller.rs
+++ b/crates/adapters/src/controller.rs
@@ -4434,31 +4434,36 @@ impl FtState {
                 let mut changed_inputs = HashMap::new();
                 let inputs = self.controller.status.input_status();
 
-                // Stop recording if the controller is shutting down. Avoid race with
-                // `stop`, which removes input endpoints from the pipeline. Without this
-                // check we may end up recording these endpoints in `remove_inputs`.
-                if self.controller.state() == PipelineState::Terminated {
-                    return Ok(());
-                }
-                self.input_endpoints
-                    .retain(|endpoint_id, (endpoint_name, paused)| {
-                        if let Some(endpoint) = inputs.get(endpoint_id) {
-                            let now_paused = endpoint.is_paused_by_user();
-                            if *paused != now_paused {
-                                changed_inputs.insert(endpoint_name.clone(), now_paused);
-                                *paused = now_paused;
+                // Compute the endpoint lifecycle diff (added/removed/paused since
+                // the previous step) that this step's journal entry records, but
+                // skip it while the controller is shutting down: `stop`
+                // concurrently disconnects endpoints, so the diff would spuriously
+                // record still-live endpoints as removed and replay them away.
+                //
+                // We must still journal this step's already-computed input, or a
+                // clean stop would silently drop the last step and a subsequent
+                // restart would resume from before it.
+                if self.controller.state() != PipelineState::Terminated {
+                    self.input_endpoints
+                        .retain(|endpoint_id, (endpoint_name, paused)| {
+                            if let Some(endpoint) = inputs.get(endpoint_id) {
+                                let now_paused = endpoint.is_paused_by_user();
+                                if *paused != now_paused {
+                                    changed_inputs.insert(endpoint_name.clone(), now_paused);
+                                    *paused = now_paused;
+                                }
+                                true
+                            } else {
+                                remove_inputs.insert(endpoint_name.clone());
+                                false
                             }
-                            true
-                        } else {
-                            remove_inputs.insert(endpoint_name.clone());
-                            false
-                        }
-                    });
-                for (endpoint_id, status) in inputs.iter() {
-                    self.input_endpoints.entry(*endpoint_id).or_insert_with(|| {
-                        add_inputs.insert(status.endpoint_name.clone(), status.config.clone());
-                        (status.endpoint_name.clone(), status.is_paused_by_user())
-                    });
+                        });
+                    for (endpoint_id, status) in inputs.iter() {
+                        self.input_endpoints.entry(*endpoint_id).or_insert_with(|| {
+                            add_inputs.insert(status.endpoint_name.clone(), status.config.clone());
+                            (status.endpoint_name.clone(), status.is_paused_by_user())
+                        });
+                    }
                 }
                 drop(inputs);
 
diff --git a/crates/adapters/src/transport/clock.rs b/crates/adapters/src/transport/clock.rs
index 579c9dec872..f76c0b54d1d 100644
--- a/crates/adapters/src/transport/clock.rs
+++ b/crates/adapters/src/transport/clock.rs
@@ -891,6 +891,13 @@ mod test {
     }
 
     /// After checkpoint/restart, `NOW()` resumes from the last replayed value, not the anchor.
+    ///
+    /// Regression coverage for a fault-tolerance journaling race: a step is
+    /// journaled after its circuit evaluation (so the entry can carry the
+    /// transaction id assigned there), and a `stop` landing between a step's
+    /// output being emitted and its journal write must not drop that step, or
+    /// the post-checkpoint advance is absent from the replay log and `NOW()`
+    /// regresses to the checkpoint value on restart.
     #[test]
     fn test_clock_advance_survives_checkpoint() {
         use dbsp::circuit::tokio::TOKIO;
@@ -987,6 +994,18 @@ mod test {
         let reader = controller.get_input_endpoint("now").unwrap();
         let clock_reader = reader.as_any().downcast::<super::ClockReader>().unwrap();
 
+        // Replay runs asynchronously on the circuit thread, feeding the journaled
+        // steps one at a time. Wait for it to finish before reading `NOW()`, or we
+        // could observe an intermediate value from before the last replayed step.
+        let deadline = std::time::Instant::now() + Duration::from_secs(30);
+        while controller.is_replaying() {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "replay did not complete within 30s"
+            );
+            sleep(Duration::from_millis(10));
+        }
+
         let post_replay = TOKIO.block_on(clock_reader.advance(Some(0))).unwrap();
         assert_eq!(
             post_replay, last_emitted_in_run1,