Preserve steering when returning to a long-running turn (#58998)

## Why

Switching back to a running thread can lose the active turn's state after its `TurnStarted` notification falls out of the bounded replay buffer, breaking steering.

## What changed

Retain the evicted active turn and include it in replay snapshots when it is missing from the stored turns. Clear the retained turn when turns are replaced, the active turn completes, or the thread closes.

## Testing

Add a regression test that evicts the turn-start notification, restores the thread, and submits two steering messages while preserving a queued follow-up. A snapshot verifies the steering messages and queued input remain visible.

GitOrigin-RevId: a4302a2e0990b907207c229361cbcec68ebbbdc4

codex-rs/tui/src/app/tests/buffered_replay.rs

@@ -599,6 +599,53 @@ async fn evicted_voice_marker_survives_widget_snapshot_for_late_reasoning() {
     }
 }
 
+#[tokio::test]
+async fn switching_back_to_a_long_running_turn_preserves_steering() {
+    let (mut app, _events, mut ops) = make_test_app_with_channels().await;
+    let (mut before_switch, _tx, _rx, _ops) = make_chatwidget_manual_with_sender().await;
+    let thread_id = ThreadId::new();
+    let session = test_thread_session(thread_id, app.config.cwd.to_path_buf());
+    before_switch.handle_thread_session(session.clone());
+    let mut store = ThreadEventStore::new_with_session(
+        /*capacity*/ 1,
+        session,
+        vec![test_turn("previous", TurnStatus::Completed, Vec::new())],
+    );
+    let started = ServerNotification::TurnStarted(TurnStartedNotification {
+        thread_id: thread_id.to_string(),
+        turn: test_turn("running", TurnStatus::InProgress, Vec::new()),
+    });
+    store.push_notification_ref(&started);
+    before_switch.handle_server_notification(started, /*replay_kind*/ None);
+    before_switch.apply_external_edit("follow-up".to_string());
+    before_switch.handle_key_event(KeyEvent::from(KeyCode::Tab));
+    store.input_state = before_switch.capture_thread_input_state();
+    store.push_notification(agent_message_delta_notification(
+        thread_id, "running", "answer", "Working",
+    ));
+    app.replay_thread_snapshot(store.snapshot(), /*resume_restored_queue*/ true);
+
+    for text in ["first steer", "second steer"] {
+        app.chat_widget.apply_external_edit(text.to_string());
+        app.chat_widget
+            .handle_key_event(KeyEvent::from(KeyCode::Enter));
+        let Op::UserTurn { items, .. } = next_user_turn_op(&mut ops) else {
+            unreachable!()
+        };
+        assert_eq!(
+            items,
+            [UserInput::Text {
+                text: text.to_string(),
+                text_elements: Vec::new(),
+            }]
+        );
+    }
+    insta::assert_snapshot!(
+        "steers_after_returning_to_a_running_turn",
+        normalize_snapshot_paths(render_bottom_popup(&app.chat_widget, /*width*/ 80))
+    );
+}
+
 #[tokio::test]
 async fn buffered_replay_renders_completed_text_without_streaming_again() {
     let (mut app, mut events, _ops) = make_test_app_with_channels().await;

codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__steers_after_returning_to_a_running_turn.snap

@@ -0,0 +1,18 @@
+---
+source: tui/src/app/tests/buffered_replay.rs
+expression: "normalize_snapshot_paths(render_bottom_popup(&app.chat_widget, 80))"
+---
+• Working
+
+• Messages to be submitted after next tool call (press esc to interrupt and send
+  immediately)
+  ↳ first steer
+  ↳ second steer
+
+• Queued follow-up inputs
+  ↳ follow-up
+    ⇧← edit last queued message
+
+› Ask Codex to do anything
+
+  gpt-test default · /tmp/project

codex-rs/tui/src/app/thread_event_buffer.rs

@@ -64,6 +64,10 @@ impl ThreadEventStore {
                         self.buffered_agent_message_delta_bytes = self
                             .buffered_agent_message_delta_bytes
                             .saturating_sub(delta.delta.len());
+                    } else if let ServerNotification::TurnStarted(started) = *notification
+                        && self.active_turn_id() == Some(&started.turn.id)
+                    {
+                        self.evicted_active_turn = Some(Box::new(started.turn));
                     }
                 }
                 ThreadBufferedEvent::Request(request) => self

codex-rs/tui/src/app/thread_events.rs

@@ -72,6 +72,8 @@ pub(super) struct ThreadEventStore {
     pub(super) buffer: VecDeque<ThreadBufferedEvent>,
     pub(super) pending_interactive_replay: PendingInteractiveReplayState,
     pub(super) active_turn_id: Option<String>,
+    // Replay still needs the current turn's start after it falls out of the event buffer.
+    pub(super) evicted_active_turn: Option<Box<Turn>>,
     // Retain the active item even if its start falls out of the bounded replay buffer.
     pub(super) active_reasoning_item: Option<codex_app_server_protocol::ItemStartedNotification>,
     // Lifecycle identity must survive bounded replay-buffer eviction.
@@ -121,6 +123,7 @@ impl ThreadEventStore {
             buffer: VecDeque::new(),
             pending_interactive_replay: PendingInteractiveReplayState::default(),
             active_turn_id: None,
+            evicted_active_turn: None,
             active_reasoning_item: None,
             latest_turn_id: None,
             pending_interrupt_turn_id: None,
@@ -157,6 +160,7 @@ impl ThreadEventStore {
     }
 
     pub(super) fn set_turns(&mut self, turns: Vec<Turn>) {
+        self.evicted_active_turn = None;
         if self.active_reasoning_item.as_ref().is_some_and(|started| {
             turns.iter().any(|turn| {
                 (turn.id == started.turn_id && turn.status != TurnStatus::InProgress)
@@ -230,7 +234,7 @@ impl ThreadEventStore {
                     self.recap_progress.completed_turns += 1;
                 }
                 if self.active_turn_id.as_deref() == Some(turn.turn.id.as_str()) {
-                    self.active_turn_id = None;
+                    self.clear_active_turn_id();
                 }
                 if self.pending_interrupt_turn_id.as_deref() == Some(turn.turn.id.as_str()) {
                     self.pending_interrupt_turn_id = None;
@@ -245,7 +249,7 @@ impl ThreadEventStore {
                 self.latest_turn_id = Some(n.turn_id.clone());
             }
             ServerNotification::ThreadClosed(_) => {
-                self.active_turn_id = None;
+                self.clear_active_turn_id();
                 self.active_reasoning_item = None;
                 self.pending_interrupt_turn_id = None;
             }
@@ -393,6 +397,12 @@ impl ThreadEventStore {
             active_reasoning_item: self.active_reasoning_item.clone(),
             input_state: self.input_state.clone(),
         };
+        if let Some(started) = self.evicted_active_turn.as_deref()
+            && self.active_turn_id() == Some(started.id.as_str())
+            && !snapshot.turns.iter().any(|turn| turn.id == started.id)
+        {
+            snapshot.turns.push(started.clone());
+        }
         if let Some(latest_turn_id) = &self.latest_turn_id {
             replay_filter::omit_resolved_misalignment_errors(&mut snapshot, latest_turn_id);
         }
@@ -549,6 +559,7 @@ impl ThreadEventStore {
 
     pub(super) fn clear_active_turn_id(&mut self) {
         self.active_turn_id = None;
+        self.evicted_active_turn = None;
     }
 }