Preserve in-flight state when restoring thread input (#33203)

## What changed

- Add an explicit restore mode for thread input snapshots and use it when replaying thread snapshots.
- Preserve the pending-steer interrupt flag alongside running-turn and queued-input state when restoring an in-flight turn.
- When in-flight state is not preserved, clear stale running flags and move pending steers ahead of already queued follow-ups while retaining their history records.

## Testing

- Expand TUI tests to cover both restoration modes, including sleep-inhibitor state, pending-steer ordering, history records, and queued submission.

GitOrigin-RevId: f4e8b2bdf065ef6c6644b7f0ea9cf32c8a9af619

codex-rs/tui/src/chatwidget/tests/composer_submission.rs

@@ -1068,38 +1068,101 @@ async fn pending_steer_interrupt_uses_remapped_binding() {
 }
 
 #[tokio::test]
-async fn restore_thread_input_state_syncs_sleep_inhibitor_state() {
-    let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
+async fn restore_thread_input_state_applies_running_state_policy() {
+    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
+    chat.thread_id = Some(ThreadId::new());
     chat.set_feature_enabled(Feature::PreventIdleSleep, /*enabled*/ true);
 
-    chat.restore_thread_input_state(Some(ThreadInputState {
-        composer: None,
+    let pending_history = UserMessageHistoryRecord::Override(UserMessageHistoryOverride {
+        text: "submitted history".to_string(),
+        text_elements: Vec::new(),
+    });
+    let queued_history = UserMessageHistoryRecord::Override(UserMessageHistoryOverride {
+        text: "queued history".to_string(),
+        text_elements: Vec::new(),
+    });
+    let input_state = ThreadInputState {
+        composer: Some(ThreadComposerState {
+            text: "composer draft".to_string(),
+            ..Default::default()
+        }),
         safety_buffering_prompt: Some(UserMessage::from("buffered prompt")),
-        pending_steers: VecDeque::new(),
-        pending_steer_history_records: VecDeque::new(),
+        pending_steers: VecDeque::from([UserMessage::from("submitted to the interrupted turn")]),
+        pending_steer_history_records: VecDeque::from([pending_history.clone()]),
         pending_steer_compare_keys: VecDeque::new(),
         rejected_steers_queue: VecDeque::new(),
         rejected_steer_history_records: VecDeque::new(),
-        queued_user_messages: VecDeque::new(),
-        queued_user_message_history_records: VecDeque::new(),
-        user_turn_pending_start: false,
+        queued_user_messages: VecDeque::from([UserMessage::from("already queued").into()]),
+        queued_user_message_history_records: VecDeque::from([queued_history.clone()]),
+        user_turn_pending_start: true,
+        submit_pending_steers_after_interrupt: true,
         current_collaboration_mode: chat.current_collaboration_mode.clone(),
         active_collaboration_mask: chat.active_collaboration_mask.clone(),
         task_running: true,
         agent_turn_running: true,
-    }));
+    };
+    chat.restore_thread_input_state(
+        Some(input_state.clone()),
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
 
     assert!(chat.turn_lifecycle.agent_turn_running);
     assert!(chat.turn_lifecycle.sleep_inhibitor.is_turn_running());
     assert!(chat.bottom_pane.is_task_running());
+    assert!(chat.input_queue.user_turn_pending_start);
+    assert!(chat.input_queue.submit_pending_steers_after_interrupt);
+    let captured_input_state = chat
+        .capture_thread_input_state()
+        .expect("thread input state");
+    assert!(captured_input_state.submit_pending_steers_after_interrupt);
     assert_eq!(
-        chat.capture_thread_input_state()
-            .expect("thread input state")
-            .safety_buffering_prompt,
+        captured_input_state.safety_buffering_prompt,
+        Some(UserMessage::from("buffered prompt"))
+    );
+    assert_eq!(chat.input_queue.pending_steers.len(), 1);
+    assert_eq!(
+        chat.safety_buffering_prompt,
         Some(UserMessage::from("buffered prompt"))
     );
 
-    chat.restore_thread_input_state(/*input_state*/ None);
+    chat.restore_thread_input_state(
+        Some(input_state),
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: false,
+        },
+    );
+
+    assert!(!chat.turn_lifecycle.agent_turn_running);
+    assert!(!chat.turn_lifecycle.sleep_inhibitor.is_turn_running());
+    assert!(!chat.bottom_pane.is_task_running());
+    assert!(!chat.input_queue.user_turn_pending_start);
+    assert!(!chat.input_queue.submit_pending_steers_after_interrupt);
+    assert!(chat.input_queue.pending_steers.is_empty());
+    assert_eq!(chat.bottom_pane.composer_text(), "composer draft");
+    assert_eq!(
+        chat.safety_buffering_prompt,
+        Some(UserMessage::from("buffered prompt"))
+    );
+    assert_eq!(
+        chat.queued_user_message_texts(),
+        vec!["submitted to the interrupted turn", "already queued"]
+    );
+    assert_eq!(
+        chat.input_queue.queued_user_message_history_records,
+        VecDeque::from([pending_history, queued_history])
+    );
+    assert!(chat.maybe_send_next_queued_input());
+    assert_matches!(next_submit_op(&mut op_rx), Op::UserTurn { .. });
+    assert_eq!(chat.queued_user_message_texts(), vec!["already queued"]);
+
+    chat.restore_thread_input_state(
+        /*input_state*/ None,
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
 
     assert!(!chat.turn_lifecycle.agent_turn_running);
     assert!(!chat.turn_lifecycle.sleep_inhibitor.is_turn_running());

codex-rs/tui/src/chatwidget/tests/review_mode.rs

@@ -373,22 +373,28 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_
     let mut queued_user_messages = VecDeque::new();
     queued_user_messages.push_back(UserMessage::from("queued draft").into());
 
-    chat.restore_thread_input_state(Some(ThreadInputState {
-        composer: None,
-        safety_buffering_prompt: None,
-        pending_steers,
-        pending_steer_history_records: VecDeque::new(),
-        pending_steer_compare_keys,
-        rejected_steers_queue,
-        rejected_steer_history_records: VecDeque::new(),
-        queued_user_messages,
-        queued_user_message_history_records: VecDeque::new(),
-        user_turn_pending_start: false,
-        current_collaboration_mode: chat.current_collaboration_mode.clone(),
-        active_collaboration_mask: chat.active_collaboration_mask.clone(),
-        task_running: false,
-        agent_turn_running: false,
-    }));
+    chat.restore_thread_input_state(
+        Some(ThreadInputState {
+            composer: None,
+            safety_buffering_prompt: None,
+            pending_steers,
+            pending_steer_history_records: VecDeque::new(),
+            pending_steer_compare_keys,
+            rejected_steers_queue,
+            rejected_steer_history_records: VecDeque::new(),
+            queued_user_messages,
+            queued_user_message_history_records: VecDeque::new(),
+            user_turn_pending_start: false,
+            submit_pending_steers_after_interrupt: false,
+            current_collaboration_mode: chat.current_collaboration_mode.clone(),
+            active_collaboration_mask: chat.active_collaboration_mask.clone(),
+            task_running: false,
+            agent_turn_running: false,
+        }),
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
 
     assert_eq!(
         chat.queued_user_message_texts(),

codex-rs/tui/src/chatwidget/tests/slash_commands.rs

@@ -539,9 +539,19 @@ async fn queued_inline_rename_does_not_drain_again_before_turn_started() {
     );
     let input_state = chat.capture_thread_input_state().unwrap();
     assert!(input_state.user_turn_pending_start);
-    chat.restore_thread_input_state(/*input_state*/ None);
+    chat.restore_thread_input_state(
+        /*input_state*/ None,
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
     assert!(!chat.input_queue.user_turn_pending_start);
-    chat.restore_thread_input_state(Some(input_state));
+    chat.restore_thread_input_state(
+        Some(input_state),
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
     assert!(chat.input_queue.user_turn_pending_start);
     assert_eq!(
         chat.queued_user_message_texts(),
@@ -987,7 +997,12 @@ async fn restored_queued_goal_slash_command_emits_set_goal_event() {
     let (mut restored_chat, mut restored_rx, mut restored_op_rx) =
         make_chatwidget_manual(/*model_override*/ None).await;
     restored_chat.set_feature_enabled(Feature::Goals, /*enabled*/ true);
-    restored_chat.restore_thread_input_state(Some(input_state));
+    restored_chat.restore_thread_input_state(
+        Some(input_state),
+        ThreadInputStateRestoreMode {
+            preserve_in_flight_turn: true,
+        },
+    );
     let thread_id = ThreadId::new();
     restored_chat.thread_id = Some(thread_id);
     restored_chat.maybe_send_next_queued_input();