diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs
index 3a28a3e83..7736aa128 100644
--- a/codex-rs/core/src/codex.rs
+++ b/codex-rs/core/src/codex.rs
@@ -6590,6 +6590,7 @@ mod tests {
     use crate::protocol::TokenCountEvent;
     use crate::protocol::TokenUsage;
     use crate::protocol::TokenUsageInfo;
+    use crate::protocol::TurnCompleteEvent;
     use crate::protocol::UserMessageEvent;
     use crate::rollout::policy::EventPersistenceMode;
     use crate::rollout::recorder::RolloutRecorder;
@@ -9307,8 +9308,8 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn task_finish_persists_leftover_pending_input() {
-        let (sess, tc, _rx) = make_session_and_context_with_rx().await;
+    async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() {
+        let (sess, tc, rx) = make_session_and_context_with_rx().await;
         let input = vec![UserInput::Text {
             text: "hello".to_string(),
             text_elements: Vec::new(),
@@ -9323,6 +9324,8 @@ mod tests {
         )
         .await;
 
+        while rx.try_recv().is_ok() {}
+
         sess.inject_response_items(vec![ResponseInputItem::Message {
             role: "user".to_string(),
             content: vec![ContentItem::InputText {
@@ -9348,6 +9351,71 @@ mod tests {
             history.raw_items().iter().any(|item| item == &expected),
             "expected pending input to be persisted into history on turn completion"
         );
+
+        let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("expected raw response item event")
+            .expect("channel open");
+        assert!(matches!(first.msg, EventMsg::RawResponseItem(_)));
+
+        let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("expected item started event")
+            .expect("channel open");
+        assert!(matches!(
+            second.msg,
+            EventMsg::ItemStarted(ItemStartedEvent {
+                item: TurnItem::UserMessage(UserMessageItem { content, .. }),
+                ..
+            }) if content == vec![UserInput::Text {
+                text: "late pending input".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ));
+
+        let third = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("expected item completed event")
+            .expect("channel open");
+        assert!(matches!(
+            third.msg,
+            EventMsg::ItemCompleted(ItemCompletedEvent {
+                item: TurnItem::UserMessage(UserMessageItem { content, .. }),
+                ..
+            }) if content == vec![UserInput::Text {
+                text: "late pending input".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ));
+
+        let fourth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("expected legacy user message event")
+            .expect("channel open");
+        assert!(matches!(
+            fourth.msg,
+            EventMsg::UserMessage(UserMessageEvent {
+                message,
+                images,
+                text_elements,
+                local_images,
+            }) if message == "late pending input"
+                && images == Some(Vec::new())
+                && text_elements.is_empty()
+                && local_images.is_empty()
+        ));
+
+        let fifth = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
+            .await
+            .expect("expected turn complete event")
+            .expect("channel open");
+        assert!(matches!(
+            fifth.msg,
+            EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id,
+                last_agent_message: None,
+            }) if turn_id == tc.sub_id
+        ));
     }
 
     #[tokio::test]
diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs
index 97719f104..5133dc1e2 100644
--- a/codex-rs/core/src/tasks/mod.rs
+++ b/codex-rs/core/src/tasks/mod.rs
@@ -22,6 +22,7 @@ use crate::AuthManager;
 use crate::codex::Session;
 use crate::codex::TurnContext;
 use crate::contextual_user_message::TURN_ABORTED_OPEN_TAG;
+use crate::event_mapping::parse_turn_item;
 use crate::models_manager::manager::ModelsManager;
 use crate::protocol::EventMsg;
 use crate::protocol::TurnAbortReason;
@@ -30,6 +31,7 @@ use crate::protocol::TurnCompleteEvent;
 use crate::state::ActiveTurn;
 use crate::state::RunningTask;
 use crate::state::TaskKind;
+use codex_protocol::items::TurnItem;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::models::ResponseItem;
@@ -213,8 +215,25 @@ impl Session {
                 .into_iter()
                 .map(ResponseItem::from)
                 .collect::<Vec<_>>();
-            self.record_conversation_items(turn_context.as_ref(), &pending_response_items)
-                .await;
+            for response_item in pending_response_items {
+                if let Some(TurnItem::UserMessage(user_message)) = parse_turn_item(&response_item) {
+                    // Keep leftover user input on the same persistence + lifecycle path as the
+                    // normal pre-sampling drain. This helper records the response item once, then
+                    // emits ItemStarted/UserMessage and ItemCompleted/UserMessage for clients.
+                    self.record_user_prompt_and_emit_turn_item(
+                        turn_context.as_ref(),
+                        &user_message.content,
+                        response_item,
+                    )
+                    .await;
+                } else {
+                    self.record_conversation_items(
+                        turn_context.as_ref(),
+                        std::slice::from_ref(&response_item),
+                    )
+                    .await;
+                }
+            }
         }
         let event = EventMsg::TurnComplete(TurnCompleteEvent {
             turn_id: turn_context.sub_id.clone(),
diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs
index 02c7faeeb..74862df99 100644
--- a/codex-rs/tui/src/bottom_pane/mod.rs
+++ b/codex-rs/tui/src/bottom_pane/mod.rs
@@ -17,8 +17,8 @@ use std::path::PathBuf;
 
 use crate::app_event::ConnectorsSnapshot;
 use crate::app_event_sender::AppEventSender;
+use crate::bottom_pane::pending_input_preview::PendingInputPreview;
 use crate::bottom_pane::pending_thread_approvals::PendingThreadApprovals;
-use crate::bottom_pane::queued_user_messages::QueuedUserMessages;
 use crate::bottom_pane::unified_exec_footer::UnifiedExecFooter;
 use crate::key_hint;
 use crate::key_hint::KeyBinding;
@@ -93,9 +93,9 @@ pub(crate) use skills_toggle_view::SkillsToggleView;
 pub(crate) use status_line_setup::StatusLineItem;
 pub(crate) use status_line_setup::StatusLineSetupView;
 mod paste_burst;
+mod pending_input_preview;
 mod pending_thread_approvals;
 pub mod popup_consts;
-mod queued_user_messages;
 mod scroll_state;
 mod selection_popup_common;
 mod textarea;
@@ -172,8 +172,8 @@ pub(crate) struct BottomPane {
     /// When a status row exists, this summary is mirrored inline in that row;
     /// when no status row exists, it renders as its own footer row.
     unified_exec_footer: UnifiedExecFooter,
-    /// Queued user messages to show above the composer while a turn is running.
-    queued_user_messages: QueuedUserMessages,
+    /// Preview of pending steers and queued drafts shown above the composer.
+    pending_input_preview: PendingInputPreview,
     /// Inactive threads with pending approval requests.
     pending_thread_approvals: PendingThreadApprovals,
     context_window_percent: Option<i64>,
@@ -223,7 +223,7 @@ impl BottomPane {
             is_task_running: false,
             status: None,
             unified_exec_footer: UnifiedExecFooter::new(),
-            queued_user_messages: QueuedUserMessages::new(),
+            pending_input_preview: PendingInputPreview::new(),
             pending_thread_approvals: PendingThreadApprovals::new(),
             esc_backtrack_hint: false,
             animations_enabled,
@@ -317,7 +317,7 @@ impl BottomPane {
     /// Update the key hint shown next to queued messages so it matches the
     /// binding that `ChatWidget` actually listens for.
     pub(crate) fn set_queued_message_edit_binding(&mut self, binding: KeyBinding) {
-        self.queued_user_messages.set_edit_binding(binding);
+        self.pending_input_preview.set_edit_binding(binding);
         self.request_redraw();
     }
 
@@ -774,9 +774,14 @@ impl BottomPane {
         true
     }
 
-    /// Update the queued messages preview shown above the composer.
-    pub(crate) fn set_queued_user_messages(&mut self, queued: Vec<String>) {
-        self.queued_user_messages.messages = queued;
+    /// Update the pending-input preview shown above the composer.
+    pub(crate) fn set_pending_input_preview(
+        &mut self,
+        queued: Vec<String>,
+        pending_steers: Vec<String>,
+    ) {
+        self.pending_input_preview.pending_steers = pending_steers;
+        self.pending_input_preview.queued_messages = queued;
         self.request_redraw();
     }
 
@@ -1019,18 +1024,19 @@ impl BottomPane {
                 flex.push(0, RenderableItem::Borrowed(&self.unified_exec_footer));
             }
             let has_pending_thread_approvals = !self.pending_thread_approvals.is_empty();
-            let has_queued_messages = !self.queued_user_messages.messages.is_empty();
+            let has_pending_input = !self.pending_input_preview.queued_messages.is_empty()
+                || !self.pending_input_preview.pending_steers.is_empty();
             let has_status_or_footer =
                 self.status.is_some() || !self.unified_exec_footer.is_empty();
-            let has_inline_previews = has_pending_thread_approvals || has_queued_messages;
+            let has_inline_previews = has_pending_thread_approvals || has_pending_input;
             if has_inline_previews && has_status_or_footer {
                 flex.push(0, RenderableItem::Owned("".into()));
             }
             flex.push(1, RenderableItem::Borrowed(&self.pending_thread_approvals));
-            if has_pending_thread_approvals && has_queued_messages {
+            if has_pending_thread_approvals && has_pending_input {
                 flex.push(0, RenderableItem::Owned("".into()));
             }
-            flex.push(1, RenderableItem::Borrowed(&self.queued_user_messages));
+            flex.push(1, RenderableItem::Borrowed(&self.pending_input_preview));
             if !has_inline_previews && has_status_or_footer {
                 flex.push(0, RenderableItem::Owned("".into()));
             }
@@ -1406,7 +1412,7 @@ mod tests {
             StatusDetailsCapitalization::CapitalizeFirst,
             STATUS_DETAILS_DEFAULT_MAX_LINES,
         );
-        pane.set_queued_user_messages(vec!["Queued follow-up question".to_string()]);
+        pane.set_pending_input_preview(vec!["Queued follow-up question".to_string()], Vec::new());
 
         let width = 48;
         let height = pane.desired_height(width);
@@ -1433,7 +1439,7 @@ mod tests {
         });
 
         pane.set_task_running(true);
-        pane.set_queued_user_messages(vec!["Queued follow-up question".to_string()]);
+        pane.set_pending_input_preview(vec!["Queued follow-up question".to_string()], Vec::new());
         pane.hide_status_indicator();
 
         let width = 48;
@@ -1461,7 +1467,7 @@ mod tests {
         });
 
         pane.set_task_running(true);
-        pane.set_queued_user_messages(vec!["Queued follow-up question".to_string()]);
+        pane.set_pending_input_preview(vec!["Queued follow-up question".to_string()], Vec::new());
 
         let width = 48;
         let height = pane.desired_height(width);
diff --git a/codex-rs/tui/src/bottom_pane/queued_user_messages.rs b/codex-rs/tui/src/bottom_pane/pending_input_preview.rs
similarity index 55%
rename from codex-rs/tui/src/bottom_pane/queued_user_messages.rs
rename to codex-rs/tui/src/bottom_pane/pending_input_preview.rs
index 30a525f45..da7bc8cb6 100644
--- a/codex-rs/tui/src/bottom_pane/queued_user_messages.rs
+++ b/codex-rs/tui/src/bottom_pane/pending_input_preview.rs
@@ -10,23 +10,26 @@ use crate::render::renderable::Renderable;
 use crate::wrapping::RtOptions;
 use crate::wrapping::adaptive_wrap_lines;
 
-/// Widget that displays a list of user messages queued while a turn is in progress.
+/// Widget that displays pending steers plus user messages queued while a turn is in progress.
 ///
-/// The widget shows a key hint at the bottom (e.g. "⌥ + ↑ edit") telling the
-/// user how to pop the most recent queued message back into the composer.
-/// Because some terminals intercept certain modifier-key combinations, the
-/// displayed binding is configurable via [`set_edit_binding`](Self::set_edit_binding).
-pub(crate) struct QueuedUserMessages {
-    pub messages: Vec<String>,
+/// The widget shows pending steers first, then queued user messages. It only
+/// shows the edit hint at the bottom (e.g. "⌥ + ↑ edit") when there are actual
+/// queued user messages to pop back into the composer. Because some terminals
+/// intercept certain modifier-key combinations, the displayed binding is
+/// configurable via [`set_edit_binding`](Self::set_edit_binding).
+pub(crate) struct PendingInputPreview {
+    pub pending_steers: Vec<String>,
+    pub queued_messages: Vec<String>,
     /// Key combination rendered in the hint line.  Defaults to Alt+Up but may
     /// be overridden for terminals where that chord is unavailable.
     edit_binding: key_hint::KeyBinding,
 }
 
-impl QueuedUserMessages {
+impl PendingInputPreview {
     pub(crate) fn new() -> Self {
         Self {
-            messages: Vec::new(),
+            pending_steers: Vec::new(),
+            queued_messages: Vec::new(),
             edit_binding: key_hint::alt(KeyCode::Up),
         }
     }
@@ -39,13 +42,31 @@ impl QueuedUserMessages {
     }
 
     fn as_renderable(&self, width: u16) -> Box<dyn Renderable> {
-        if self.messages.is_empty() || width < 4 {
+        if (self.pending_steers.is_empty() && self.queued_messages.is_empty()) || width < 4 {
             return Box::new(());
         }
 
         let mut lines = vec![];
 
-        for message in &self.messages {
+        for steer in &self.pending_steers {
+            let wrapped = adaptive_wrap_lines(
+                steer
+                    .lines()
+                    .map(|line| format!("pending steer: {line}").dim()),
+                RtOptions::new(width as usize)
+                    .initial_indent(Line::from("  ! ".dim()))
+                    .subsequent_indent(Line::from("    ")),
+            );
+            let len = wrapped.len();
+            for line in wrapped.into_iter().take(3) {
+                lines.push(line);
+            }
+            if len > 3 {
+                lines.push(Line::from("    …".dim()));
+            }
+        }
+
+        for message in &self.queued_messages {
             let wrapped = adaptive_wrap_lines(
                 message.lines().map(|line| line.dim().italic()),
                 RtOptions::new(width as usize)
@@ -61,20 +82,22 @@ impl QueuedUserMessages {
             }
         }
 
-        lines.push(
-            Line::from(vec![
-                "    ".into(),
-                self.edit_binding.into(),
-                " edit".into(),
-            ])
-            .dim(),
-        );
+        if !self.queued_messages.is_empty() {
+            lines.push(
+                Line::from(vec![
+                    "    ".into(),
+                    self.edit_binding.into(),
+                    " edit".into(),
+                ])
+                .dim(),
+            );
+        }
 
         Paragraph::new(lines).into()
     }
 }
 
-impl Renderable for QueuedUserMessages {
+impl Renderable for PendingInputPreview {
     fn render(&self, area: Rect, buf: &mut Buffer) {
         if area.is_empty() {
             return;
@@ -96,21 +119,21 @@ mod tests {
 
     #[test]
     fn desired_height_empty() {
-        let queue = QueuedUserMessages::new();
+        let queue = PendingInputPreview::new();
         assert_eq!(queue.desired_height(40), 0);
     }
 
     #[test]
     fn desired_height_one_message() {
-        let mut queue = QueuedUserMessages::new();
-        queue.messages.push("Hello, world!".to_string());
+        let mut queue = PendingInputPreview::new();
+        queue.queued_messages.push("Hello, world!".to_string());
         assert_eq!(queue.desired_height(40), 2);
     }
 
     #[test]
     fn render_one_message() {
-        let mut queue = QueuedUserMessages::new();
-        queue.messages.push("Hello, world!".to_string());
+        let mut queue = PendingInputPreview::new();
+        queue.queued_messages.push("Hello, world!".to_string());
         let width = 40;
         let height = queue.desired_height(width);
         let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
@@ -120,9 +143,11 @@ mod tests {
 
     #[test]
     fn render_two_messages() {
-        let mut queue = QueuedUserMessages::new();
-        queue.messages.push("Hello, world!".to_string());
-        queue.messages.push("This is another message".to_string());
+        let mut queue = PendingInputPreview::new();
+        queue.queued_messages.push("Hello, world!".to_string());
+        queue
+            .queued_messages
+            .push("This is another message".to_string());
         let width = 40;
         let height = queue.desired_height(width);
         let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
@@ -132,11 +157,17 @@ mod tests {
 
     #[test]
     fn render_more_than_three_messages() {
-        let mut queue = QueuedUserMessages::new();
-        queue.messages.push("Hello, world!".to_string());
-        queue.messages.push("This is another message".to_string());
-        queue.messages.push("This is a third message".to_string());
-        queue.messages.push("This is a fourth message".to_string());
+        let mut queue = PendingInputPreview::new();
+        queue.queued_messages.push("Hello, world!".to_string());
+        queue
+            .queued_messages
+            .push("This is another message".to_string());
+        queue
+            .queued_messages
+            .push("This is a third message".to_string());
+        queue
+            .queued_messages
+            .push("This is a fourth message".to_string());
         let width = 40;
         let height = queue.desired_height(width);
         let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
@@ -146,11 +177,13 @@ mod tests {
 
     #[test]
     fn render_wrapped_message() {
-        let mut queue = QueuedUserMessages::new();
+        let mut queue = PendingInputPreview::new();
         queue
-            .messages
+            .queued_messages
             .push("This is a longer message that should be wrapped".to_string());
-        queue.messages.push("This is another message".to_string());
+        queue
+            .queued_messages
+            .push("This is another message".to_string());
         let width = 40;
         let height = queue.desired_height(width);
         let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
@@ -160,9 +193,9 @@ mod tests {
 
     #[test]
     fn render_many_line_message() {
-        let mut queue = QueuedUserMessages::new();
+        let mut queue = PendingInputPreview::new();
         queue
-            .messages
+            .queued_messages
             .push("This is\na message\nwith many\nlines".to_string());
         let width = 40;
         let height = queue.desired_height(width);
@@ -173,8 +206,8 @@ mod tests {
 
     #[test]
     fn long_url_like_message_does_not_expand_into_wrapped_ellipsis_rows() {
-        let mut queue = QueuedUserMessages::new();
-        queue.messages.push(
+        let mut queue = PendingInputPreview::new();
+        queue.queued_messages.push(
             "example.test/api/v1/projects/alpha-team/releases/2026-02-17/builds/1234567890/artifacts/reports/performance/summary/detail/session_id=abc123def456ghi789"
                 .to_string(),
         );
@@ -202,4 +235,35 @@ mod tests {
             "expected no wrapped-ellipsis row for URL-like token, got rows: {rendered_rows:?}"
         );
     }
+
+    #[test]
+    fn render_one_pending_steer() {
+        let mut queue = PendingInputPreview::new();
+        queue.pending_steers.push("Please continue.".to_string());
+        let width = 48;
+        let height = queue.desired_height(width);
+        let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
+        queue.render(Rect::new(0, 0, width, height), &mut buf);
+        assert_snapshot!("render_one_pending_steer", format!("{buf:?}"));
+    }
+
+    #[test]
+    fn render_pending_steers_above_queued_messages() {
+        let mut queue = PendingInputPreview::new();
+        queue.pending_steers.push("Please continue.".to_string());
+        queue
+            .pending_steers
+            .push("Check the last command output.".to_string());
+        queue
+            .queued_messages
+            .push("Queued follow-up question".to_string());
+        let width = 52;
+        let height = queue.desired_height(width);
+        let mut buf = Buffer::empty(Rect::new(0, 0, width, height));
+        queue.render(Rect::new(0, 0, width, height), &mut buf);
+        assert_snapshot!(
+            "render_pending_steers_above_queued_messages",
+            format!("{buf:?}")
+        );
+    }
 }
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_many_line_message.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_many_line_message.snap
similarity index 100%
rename from codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_many_line_message.snap
rename to codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_many_line_message.snap
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_more_than_three_messages.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_more_than_three_messages.snap
similarity index 100%
rename from codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_more_than_three_messages.snap
rename to codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_more_than_three_messages.snap
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_one_message.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_one_message.snap
similarity index 100%
rename from codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_one_message.snap
rename to codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_one_message.snap
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_one_pending_steer.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_one_pending_steer.snap
new file mode 100644
index 000000000..d738b7dc0
--- /dev/null
+++ b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_one_pending_steer.snap
@@ -0,0 +1,15 @@
+---
+source: tui/src/bottom_pane/pending_input_preview.rs
+assertion_line: 237
+expression: "format!(\"{buf:?}\")"
+---
+Buffer {
+    area: Rect { x: 0, y: 0, width: 48, height: 1 },
+    content: [
+        "  ! pending steer: Please continue.             ",
+    ],
+    styles: [
+        x: 0, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: DIM,
+        x: 35, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
+    ]
+}
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_pending_steers_above_queued_messages.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_pending_steers_above_queued_messages.snap
new file mode 100644
index 000000000..6979d507d
--- /dev/null
+++ b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_pending_steers_above_queued_messages.snap
@@ -0,0 +1,25 @@
+---
+source: tui/src/bottom_pane/pending_input_preview.rs
+assertion_line: 252
+expression: "format!(\"{buf:?}\")"
+---
+Buffer {
+    area: Rect { x: 0, y: 0, width: 52, height: 4 },
+    content: [
+        "  ! pending steer: Please continue.                 ",
+        "  ! pending steer: Check the last command output.   ",
+        "  ↳ Queued follow-up question                       ",
+        "    ⌥ + ↑ edit                                      ",
+    ],
+    styles: [
+        x: 0, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: DIM,
+        x: 35, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
+        x: 0, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: DIM,
+        x: 49, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
+        x: 0, y: 2, fg: Reset, bg: Reset, underline: Reset, modifier: DIM,
+        x: 4, y: 2, fg: Reset, bg: Reset, underline: Reset, modifier: DIM | ITALIC,
+        x: 29, y: 2, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
+        x: 0, y: 3, fg: Reset, bg: Reset, underline: Reset, modifier: DIM,
+        x: 14, y: 3, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
+    ]
+}
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_two_messages.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_two_messages.snap
similarity index 100%
rename from codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_two_messages.snap
rename to codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_two_messages.snap
diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_wrapped_message.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_wrapped_message.snap
similarity index 100%
rename from codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__queued_user_messages__tests__render_wrapped_message.snap
rename to codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_wrapped_message.snap
diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs
index 51f39c1f2..a0d345418 100644
--- a/codex-rs/tui/src/chatwidget.rs
+++ b/codex-rs/tui/src/chatwidget.rs
@@ -37,6 +37,7 @@ use std::sync::atomic::Ordering;
 use std::time::Duration;
 use std::time::Instant;
 
+use self::realtime::PendingSteerCompareKey;
 use crate::app_event::RealtimeAudioDeviceKind;
 #[cfg(all(not(target_os = "linux"), feature = "voice-input"))]
 use crate::audio_device::list_realtime_audio_device_names;
@@ -85,6 +86,7 @@ use codex_protocol::config_types::ServiceTier;
 use codex_protocol::config_types::Settings;
 #[cfg(target_os = "windows")]
 use codex_protocol::config_types::WindowsSandboxLevel;
+use codex_protocol::items::AgentMessageContent;
 use codex_protocol::items::AgentMessageItem;
 use codex_protocol::models::MessagePhase;
 use codex_protocol::models::local_image_label_text;
@@ -618,6 +620,11 @@ pub(crate) struct ChatWidget {
     suppress_session_configured_redraw: bool,
     // User messages queued while a turn is in progress
     queued_user_messages: VecDeque<UserMessage>,
+    // Steers already submitted to core but not yet committed into history.
+    //
+    // The bottom pane shows these above queued drafts until core records the
+    // corresponding user message item.
+    pending_steers: VecDeque<PendingSteer>,
     /// Terminal-appropriate keybinding for popping the most-recently queued
     /// message back into the composer.  Determined once at construction time via
     /// [`queued_message_edit_binding_for_terminal`] and propagated to
@@ -650,7 +657,9 @@ pub(crate) struct ChatWidget {
     had_work_activity: bool,
     // Whether the current turn emitted a plan update.
     saw_plan_update_this_turn: bool,
-    // Whether the current turn emitted a proposed plan item.
+    // Whether the current turn emitted a proposed plan item that has not been superseded by a
+    // later steer. This is cleared when the user submits a steer so the plan popup only appears
+    // if a newer proposed plan arrives afterward.
     saw_plan_item_this_turn: bool,
     // Incremental buffer for streamed plan content.
     plan_delta_buffer: String,
@@ -751,6 +760,11 @@ impl From<&str> for UserMessage {
     }
 }
 
+struct PendingSteer {
+    user_message: UserMessage,
+    compare_key: PendingSteerCompareKey,
+}
+
 pub(crate) fn create_initial_user_message(
     text: Option<String>,
     local_image_paths: Vec<PathBuf>,
@@ -778,6 +792,21 @@ pub(crate) fn create_initial_user_message(
     }
 }
 
+fn append_text_with_rebased_elements(
+    target_text: &mut String,
+    target_text_elements: &mut Vec<TextElement>,
+    text: &str,
+    text_elements: impl IntoIterator<Item = TextElement>,
+) {
+    let offset = target_text.len();
+    target_text.push_str(text);
+    target_text_elements.extend(text_elements.into_iter().map(|mut element| {
+        element.byte_range.start += offset;
+        element.byte_range.end += offset;
+        element
+    }));
+}
+
 // When merging multiple queued drafts (e.g., after interrupt), each draft starts numbering
 // its attachments at [Image #1]. Reassign placeholder labels based on the attachment list so
 // the combined local_image_paths order matches the labels, even if placeholders were moved
@@ -1290,17 +1319,24 @@ impl ChatWidget {
         self.request_redraw();
     }
 
-    fn on_agent_message(&mut self, message: String) {
-        // If we have a stream_controller, then the final agent message is redundant and will be a
-        // duplicate of what has already been streamed.
-        if self.stream_controller.is_none() && !message.is_empty() {
-            self.handle_streaming_delta(message);
+    fn finalize_completed_assistant_message(&mut self, message: Option<&str>) {
+        // If we have a stream_controller, the finalized message payload is redundant because the
+        // visible content has already been accumulated through deltas.
+        if self.stream_controller.is_none()
+            && let Some(message) = message
+            && !message.is_empty()
+        {
+            self.handle_streaming_delta(message.to_string());
         }
         self.flush_answer_stream_with_separator();
         self.handle_stream_finished();
         self.request_redraw();
     }
 
+    fn on_agent_message(&mut self, message: String) {
+        self.finalize_completed_assistant_message(Some(&message));
+    }
+
     fn on_agent_message_delta(&mut self, delta: String) {
         self.handle_streaming_delta(delta);
     }
@@ -1483,7 +1519,10 @@ impl ChatWidget {
         self.unified_exec_wait_streak = None;
         self.request_redraw();
 
-        if !from_replay && self.queued_user_messages.is_empty() {
+        let had_pending_steers = !self.pending_steers.is_empty();
+        self.refresh_pending_input_preview();
+
+        if !from_replay && self.queued_user_messages.is_empty() && !had_pending_steers {
             self.maybe_prompt_plan_implementation();
         }
         // Keep this flag for replayed completion events so a subsequent live TurnComplete can
@@ -1865,30 +1904,31 @@ impl ChatWidget {
         if reason == TurnAbortReason::Interrupted {
             self.clear_unified_exec_processes();
         }
-
         if reason != TurnAbortReason::ReviewEnded {
             self.add_to_history(history_cell::new_error_event(
                 "Conversation interrupted - tell the model what to do differently. Something went wrong? Hit `/feedback` to report the issue.".to_owned(),
             ));
         }
 
-        if let Some(combined) = self.drain_queued_messages_for_restore() {
+        // Core clears pending_input before emitting TurnAborted, so any unacknowledged steers
+        // still tracked here must be restored locally instead of waiting for a later commit.
+        if let Some(combined) = self.drain_pending_messages_for_restore() {
             self.restore_user_message_to_composer(combined);
-            self.refresh_queued_user_messages();
         }
+        self.refresh_pending_input_preview();
 
         self.request_redraw();
     }
 
-    /// Merge queued drafts (plus the current composer state) into a single message for restore.
+    /// Merge pending steers, queued drafts, and the current composer state into a single message.
     ///
-    /// Each queued draft numbers attachments from `[Image #1]`. When we concatenate drafts, we
-    /// must renumber placeholders in a stable order so the merged attachment list stays aligned
-    /// with the labels embedded in text. This helper drains the queue, remaps placeholders, and
-    /// fixes text element byte ranges as content is appended. Returns `None` when there is nothing
-    /// to restore.
-    fn drain_queued_messages_for_restore(&mut self) -> Option<UserMessage> {
-        if self.queued_user_messages.is_empty() {
+    /// Each pending message numbers attachments from `[Image #1]` relative to its own remote
+    /// images. When we concatenate multiple messages after interrupt, we must renumber local-image
+    /// placeholders in a stable order and rebase text element byte ranges so the restored composer
+    /// state stays aligned with the merged attachment list. Returns `None` when there is nothing to
+    /// restore.
+    fn drain_pending_messages_for_restore(&mut self) -> Option<UserMessage> {
+        if self.pending_steers.is_empty() && self.queued_user_messages.is_empty() {
             return None;
         }
 
@@ -1900,7 +1940,12 @@ impl ChatWidget {
             mention_bindings: self.bottom_pane.composer_mention_bindings(),
         };
 
-        let mut to_merge: Vec<UserMessage> = self.queued_user_messages.drain(..).collect();
+        let mut to_merge: Vec<UserMessage> = self
+            .pending_steers
+            .drain(..)
+            .map(|steer| steer.user_message)
+            .collect();
+        to_merge.extend(self.queued_user_messages.drain(..));
         if !existing_message.text.is_empty()
             || !existing_message.local_images.is_empty()
             || !existing_message.remote_image_urls.is_empty()
@@ -1915,7 +1960,6 @@ impl ChatWidget {
             remote_image_urls: Vec::new(),
             mention_bindings: Vec::new(),
         };
-        let mut combined_offset = 0usize;
         let total_remote_images = to_merge
             .iter()
             .map(|message| message.remote_image_urls.len())
@@ -1925,22 +1969,23 @@ impl ChatWidget {
         for (idx, message) in to_merge.into_iter().enumerate() {
             if idx > 0 {
                 combined.text.push('\n');
-                combined_offset += 1;
             }
-            let message = remap_placeholders_for_message(message, &mut next_image_label);
-            let base = combined_offset;
-            combined.text.push_str(&message.text);
-            combined_offset += message.text.len();
-            combined
-                .text_elements
-                .extend(message.text_elements.into_iter().map(|mut elem| {
-                    elem.byte_range.start += base;
-                    elem.byte_range.end += base;
-                    elem
-                }));
-            combined.local_images.extend(message.local_images);
-            combined.remote_image_urls.extend(message.remote_image_urls);
-            combined.mention_bindings.extend(message.mention_bindings);
+            let UserMessage {
+                text,
+                text_elements,
+                local_images,
+                remote_image_urls,
+                mention_bindings,
+            } = remap_placeholders_for_message(message, &mut next_image_label);
+            append_text_with_rebased_elements(
+                &mut combined.text,
+                &mut combined.text_elements,
+                &text,
+                text_elements,
+            );
+            combined.local_images.extend(local_images);
+            combined.remote_image_urls.extend(remote_image_urls);
+            combined.mention_bindings.extend(mention_bindings);
         }
 
         Some(combined)
@@ -2350,6 +2395,15 @@ impl ChatWidget {
     /// returns once stream queues are idle. Final-answer completion (or absent
     /// phase for legacy models) clears the flag to preserve historical behavior.
     fn on_agent_message_item_completed(&mut self, item: AgentMessageItem) {
+        let mut message = String::new();
+        for content in &item.content {
+            match content {
+                AgentMessageContent::Text { text } => message.push_str(text),
+            }
+        }
+        self.finalize_completed_assistant_message(
+            (!message.is_empty()).then_some(message.as_str()),
+        );
         self.pending_status_indicator_restore = match item.phase {
             // Models that don't support preambles only output AgentMessageItems on turn completion.
             Some(MessagePhase::FinalAnswer) | None => false,
@@ -2915,6 +2969,7 @@ impl ChatWidget {
             thread_name: None,
             forked_from: None,
             queued_user_messages: VecDeque::new(),
+            pending_steers: VecDeque::new(),
             queued_message_edit_binding,
             show_welcome_banner: is_first_run,
             startup_tooltip_override,
@@ -3099,6 +3154,7 @@ impl ChatWidget {
             plan_delta_buffer: String::new(),
             plan_item_active: false,
             queued_user_messages: VecDeque::new(),
+            pending_steers: VecDeque::new(),
             queued_message_edit_binding,
             show_welcome_banner: is_first_run,
             startup_tooltip_override,
@@ -3264,6 +3320,7 @@ impl ChatWidget {
             thread_name: None,
             forked_from: None,
             queued_user_messages: VecDeque::new(),
+            pending_steers: VecDeque::new(),
             queued_message_edit_binding,
             show_welcome_banner: false,
             startup_tooltip_override: None,
@@ -3394,7 +3451,7 @@ impl ChatWidget {
         {
             if let Some(user_message) = self.queued_user_messages.pop_back() {
                 self.restore_user_message_to_composer(user_message);
-                self.refresh_queued_user_messages();
+                self.refresh_pending_input_preview();
                 self.request_redraw();
             }
             return;
@@ -3429,18 +3486,19 @@ impl ChatWidget {
                             .bottom_pane
                             .take_recent_submission_mention_bindings(),
                     };
+                    if user_message.text.is_empty()
+                        && user_message.local_images.is_empty()
+                        && user_message.remote_image_urls.is_empty()
+                    {
+                        return;
+                    }
                     let Some(user_message) =
                         self.maybe_defer_user_message_for_realtime(user_message)
                     else {
                         return;
                     };
-                    // Submissions during active final-answer streaming can race with turn
-                    // completion and strand the UI in a running state. Queue those inputs instead
-                    // of injecting immediately; `on_task_complete()` drains this FIFO via
-                    // `maybe_send_next_queued_input()`, so no typed prompt is dropped.
-                    let should_submit_now = self.is_session_configured()
-                        && !self.is_plan_streaming_in_tui()
-                        && self.stream_controller.is_none();
+                    let should_submit_now =
+                        self.is_session_configured() && !self.is_plan_streaming_in_tui();
                     if should_submit_now {
                         // Submitted is emitted when user submits.
                         // Reset any reasoning header only when we are actually submitting a turn.
@@ -4086,7 +4144,7 @@ impl ChatWidget {
             || self.is_review_mode
         {
             self.queued_user_messages.push_back(user_message);
-            self.refresh_queued_user_messages();
+            self.refresh_pending_input_preview();
         } else {
             self.submit_user_message(user_message);
         }
@@ -4096,7 +4154,12 @@ impl ChatWidget {
         if !self.is_session_configured() {
             tracing::warn!("cannot submit user message before session is configured; queueing");
             self.queued_user_messages.push_front(user_message);
-            self.refresh_queued_user_messages();
+            self.refresh_pending_input_preview();
+            return;
+        }
+        if self.is_review_mode {
+            self.queued_user_messages.push_back(user_message);
+            self.refresh_pending_input_preview();
             return;
         }
 
@@ -4123,6 +4186,7 @@ impl ChatWidget {
             return;
         }
 
+        let render_in_history = !self.agent_turn_running;
         let mut items: Vec<UserInput> = Vec::new();
 
         // Special-case: "!cmd" executes a local shell command instead of sending to the model.
@@ -4251,6 +4315,16 @@ impl ChatWidget {
         } else {
             None
         };
+        let pending_steer = (!render_in_history).then(|| PendingSteer {
+            user_message: UserMessage {
+                text: text.clone(),
+                local_images: local_images.clone(),
+                remote_image_urls: remote_image_urls.clone(),
+                text_elements: text_elements.clone(),
+                mention_bindings: mention_bindings.clone(),
+            },
+            compare_key: Self::pending_steer_compare_key_from_items(&items),
+        });
         let personality = self
             .config
             .personality
@@ -4271,9 +4345,9 @@ impl ChatWidget {
             personality,
         };
 
-        self.codex_op_tx.send(op).unwrap_or_else(|e| {
-            tracing::error!("failed to send message: {e}");
-        });
+        if !self.submit_op(op) {
+            return;
+        }
 
         // Persist the text to cross-session message history.
         if !text.is_empty() {
@@ -4292,8 +4366,14 @@ impl ChatWidget {
                 });
         }
 
+        if let Some(pending_steer) = pending_steer {
+            self.pending_steers.push_back(pending_steer);
+            self.saw_plan_item_this_turn = false;
+            self.refresh_pending_input_preview();
+        }
+
         // Show replayable user content in conversation history.
-        if !text.is_empty() {
+        if render_in_history && !text.is_empty() {
             let local_image_paths = local_images
                 .into_iter()
                 .map(|img| img.path)
@@ -4311,7 +4391,7 @@ impl ChatWidget {
                 local_image_paths,
                 remote_image_urls,
             ));
-        } else if !remote_image_urls.is_empty() {
+        } else if render_in_history && !remote_image_urls.is_empty() {
             self.last_rendered_user_message_event =
                 Some(Self::rendered_user_message_event_from_parts(
                     String::new(),
@@ -4424,9 +4504,18 @@ impl ChatWidget {
         match msg {
             EventMsg::SessionConfigured(e) => self.on_session_configured(e),
             EventMsg::ThreadNameUpdated(e) => self.on_thread_name_updated(e),
-            EventMsg::AgentMessage(AgentMessageEvent { message, .. }) => {
+            EventMsg::AgentMessage(AgentMessageEvent { .. })
+                if matches!(replay_kind, Some(ReplayKind::ThreadSnapshot))
+                    && !self.is_review_mode => {}
+            EventMsg::AgentMessage(AgentMessageEvent { message, .. })
+                if from_replay || self.is_review_mode =>
+            {
+                // TODO(ccunningham): stop relying on legacy AgentMessage in review mode,
+                // including thread-snapshot replay, and forward
+                // ItemCompleted(TurnItem::AgentMessage(_)) instead.
                 self.on_agent_message(message)
             }
+            EventMsg::AgentMessage(AgentMessageEvent { .. }) => {}
             EventMsg::AgentMessageDelta(AgentMessageDeltaEvent { delta }) => {
                 self.on_agent_message_delta(delta)
             }
@@ -4481,6 +4570,8 @@ impl ChatWidget {
                     self.on_interrupted_turn(ev.reason);
                 }
                 TurnAbortReason::Replaced => {
+                    self.pending_steers.clear();
+                    self.refresh_pending_input_preview();
                     self.on_error("Turn aborted: replaced by a new task".to_owned())
                 }
                 TurnAbortReason::ReviewEnded => {
@@ -4599,6 +4690,42 @@ impl ChatWidget {
             }
             EventMsg::ItemCompleted(event) => {
                 let item = event.item;
+                if !from_replay && let codex_protocol::items::TurnItem::UserMessage(item) = &item {
+                    let EventMsg::UserMessage(event) = item.as_legacy_event() else {
+                        unreachable!("user message item should convert to a legacy user message");
+                    };
+                    let rendered = Self::rendered_user_message_event_from_event(&event);
+                    let compare_key = Self::pending_steer_compare_key_from_item(item);
+                    if self
+                        .pending_steers
+                        .front()
+                        .is_some_and(|pending| pending.compare_key == compare_key)
+                    {
+                        if let Some(pending) = self.pending_steers.pop_front() {
+                            self.refresh_pending_input_preview();
+                            let pending_event = UserMessageEvent {
+                                message: pending.user_message.text,
+                                images: Some(pending.user_message.remote_image_urls),
+                                local_images: pending
+                                    .user_message
+                                    .local_images
+                                    .into_iter()
+                                    .map(|image| image.path)
+                                    .collect(),
+                                text_elements: pending.user_message.text_elements,
+                            };
+                            self.on_user_message_event(pending_event);
+                        } else if self.last_rendered_user_message_event.as_ref() != Some(&rendered)
+                        {
+                            tracing::warn!(
+                                "pending steer matched compare key but queue was empty when rendering committed user message"
+                            );
+                            self.on_user_message_event(event);
+                        }
+                    } else if self.last_rendered_user_message_event.as_ref() != Some(&rendered) {
+                        self.on_user_message_event(event);
+                    }
+                }
                 if let codex_protocol::items::TurnItem::Plan(plan_item) = &item {
                     self.on_plan_item_completed(plan_item.text.clone());
                 }
@@ -4749,17 +4876,23 @@ impl ChatWidget {
             self.submit_user_message(user_message);
         }
         // Update the list to reflect the remaining queued messages (if any).
-        self.refresh_queued_user_messages();
+        self.refresh_pending_input_preview();
     }
 
-    /// Rebuild and update the queued user messages from the current queue.
-    fn refresh_queued_user_messages(&mut self) {
-        let messages: Vec<String> = self
+    /// Rebuild and update the bottom-pane pending-input preview.
+    fn refresh_pending_input_preview(&mut self) {
+        let queued_messages: Vec<String> = self
             .queued_user_messages
             .iter()
             .map(|m| m.text.clone())
             .collect();
-        self.bottom_pane.set_queued_user_messages(messages);
+        let pending_steers: Vec<String> = self
+            .pending_steers
+            .iter()
+            .map(|steer| steer.user_message.text.clone())
+            .collect();
+        self.bottom_pane
+            .set_pending_input_preview(queued_messages, pending_steers);
     }
 
     pub(crate) fn set_pending_thread_approvals(&mut self, threads: Vec<String>) {
diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs
index 2970ac4d3..32f0a1528 100644
--- a/codex-rs/tui/src/chatwidget/realtime.rs
+++ b/codex-rs/tui/src/chatwidget/realtime.rs
@@ -49,10 +49,16 @@ impl RealtimeConversationUiState {
 
 #[derive(Clone, Debug, PartialEq)]
 pub(super) struct RenderedUserMessageEvent {
-    message: String,
-    remote_image_urls: Vec<String>,
-    local_images: Vec<PathBuf>,
-    text_elements: Vec<TextElement>,
+    pub(super) message: String,
+    pub(super) remote_image_urls: Vec<String>,
+    pub(super) local_images: Vec<PathBuf>,
+    pub(super) text_elements: Vec<TextElement>,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) struct PendingSteerCompareKey {
+    pub(super) message: String,
+    pub(super) image_count: usize,
 }
 
 impl ChatWidget {
@@ -81,6 +87,78 @@ impl ChatWidget {
         )
     }
 
+    /// Build the compare key for a submitted pending steer without invoking the
+    /// expensive request-serialization path. Pending steers only need to match the
+    /// committed `ItemCompleted(UserMessage)` emitted after core drains input, which
+    /// preserves flattened text and total image count but not UI-only text ranges or
+    /// local image paths.
+    pub(super) fn pending_steer_compare_key_from_items(
+        items: &[UserInput],
+    ) -> PendingSteerCompareKey {
+        let mut message = String::new();
+        let mut image_count = 0;
+
+        for item in items {
+            match item {
+                UserInput::Text { text, .. } => message.push_str(text),
+                UserInput::Image { .. } | UserInput::LocalImage { .. } => image_count += 1,
+                UserInput::Skill { .. } | UserInput::Mention { .. } => {}
+                _ => {}
+            }
+        }
+
+        PendingSteerCompareKey {
+            message,
+            image_count,
+        }
+    }
+
+    pub(super) fn pending_steer_compare_key_from_item(
+        item: &codex_protocol::items::UserMessageItem,
+    ) -> PendingSteerCompareKey {
+        Self::pending_steer_compare_key_from_items(&item.content)
+    }
+
+    #[cfg(test)]
+    pub(super) fn rendered_user_message_event_from_inputs(
+        items: &[UserInput],
+    ) -> RenderedUserMessageEvent {
+        let mut message = String::new();
+        let mut remote_image_urls = Vec::new();
+        let mut local_images = Vec::new();
+        let mut text_elements = Vec::new();
+
+        for item in items {
+            match item {
+                UserInput::Text {
+                    text,
+                    text_elements: current_text_elements,
+                } => append_text_with_rebased_elements(
+                    &mut message,
+                    &mut text_elements,
+                    text,
+                    current_text_elements.iter().map(|element| {
+                        TextElement::new(
+                            element.byte_range,
+                            element.placeholder(text).map(str::to_string),
+                        )
+                    }),
+                ),
+                UserInput::Image { image_url } => remote_image_urls.push(image_url.clone()),
+                UserInput::LocalImage { path } => local_images.push(path.clone()),
+                UserInput::Skill { .. } | UserInput::Mention { .. } => {}
+                _ => {}
+            }
+        }
+
+        Self::rendered_user_message_event_from_parts(
+            message,
+            text_elements,
+            local_images,
+            remote_image_urls,
+        )
+    }
+
     pub(super) fn should_render_realtime_user_message_event(
         &self,
         event: &UserMessageEvent,
diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__deltas_then_same_final_message_are_rendered_snapshot.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__deltas_then_same_final_message_are_rendered_snapshot.snap
deleted file mode 100644
index 606208718..000000000
--- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__deltas_then_same_final_message_are_rendered_snapshot.snap
+++ /dev/null
@@ -1,5 +0,0 @@
----
-source: tui/src/chatwidget/tests.rs
-expression: combined
----
-• Here is the result.
diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs
index a720559b5..6ea109dac 100644
--- a/codex-rs/tui/src/chatwidget/tests.rs
+++ b/codex-rs/tui/src/chatwidget/tests.rs
@@ -44,6 +44,7 @@ use codex_protocol::items::AgentMessageContent;
 use codex_protocol::items::AgentMessageItem;
 use codex_protocol::items::PlanItem;
 use codex_protocol::items::TurnItem;
+use codex_protocol::items::UserMessageItem;
 use codex_protocol::models::MessagePhase;
 use codex_protocol::openai_models::ModelPreset;
 use codex_protocol::openai_models::ReasoningEffortPreset;
@@ -211,6 +212,45 @@ async fn resumed_initial_messages_render_history() {
     );
 }
 
+#[tokio::test]
+async fn thread_snapshot_replay_does_not_duplicate_agent_message_history() {
+    let (mut chat, mut rx, _ops) = make_chatwidget_manual(None).await;
+
+    chat.handle_codex_event_replay(Event {
+        id: "turn-1".into(),
+        msg: EventMsg::ItemCompleted(ItemCompletedEvent {
+            thread_id: ThreadId::new(),
+            turn_id: "turn-1".to_string(),
+            item: TurnItem::AgentMessage(AgentMessageItem {
+                id: "msg-1".to_string(),
+                content: vec![AgentMessageContent::Text {
+                    text: "assistant reply".to_string(),
+                }],
+                phase: None,
+            }),
+        }),
+    });
+    chat.handle_codex_event_replay(Event {
+        id: "turn-1".into(),
+        msg: EventMsg::AgentMessage(AgentMessageEvent {
+            message: "assistant reply".to_string(),
+            phase: None,
+        }),
+    });
+
+    let cells = drain_insert_history(&mut rx);
+    assert_eq!(
+        cells.len(),
+        1,
+        "expected replayed assistant message to render once"
+    );
+    let rendered = lines_to_single_string(&cells[0]);
+    assert!(
+        rendered.contains("assistant reply"),
+        "expected replayed assistant message, got {rendered:?}"
+    );
+}
+
 #[tokio::test]
 async fn replayed_user_message_preserves_text_elements_and_local_images() {
     let (mut chat, mut rx, _ops) = make_chatwidget_manual(None).await;
@@ -1181,7 +1221,7 @@ async fn interrupted_turn_restores_queued_messages_with_images_and_elements() {
         text_elements: second_elements,
         mention_bindings: Vec::new(),
     });
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     chat.bottom_pane
         .set_composer_text(existing_text, existing_elements, existing_images.clone());
@@ -1251,7 +1291,7 @@ async fn interrupted_turn_restore_keeps_active_mode_for_resubmission() {
         text_elements: Vec::new(),
         mention_bindings: Vec::new(),
     });
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     chat.handle_codex_event(Event {
         id: "interrupt".into(),
@@ -1451,6 +1491,58 @@ async fn entered_review_mode_defaults_to_current_changes_banner() {
     assert!(chat.is_review_mode);
 }
 
+#[tokio::test]
+async fn live_agent_message_renders_during_review_mode() {
+    let (mut chat, mut rx, _ops) = make_chatwidget_manual(None).await;
+
+    chat.handle_codex_event(Event {
+        id: "review-start".into(),
+        msg: EventMsg::EnteredReviewMode(ReviewRequest {
+            target: ReviewTarget::UncommittedChanges,
+            user_facing_hint: None,
+        }),
+    });
+    let _ = drain_insert_history(&mut rx);
+
+    chat.handle_codex_event(Event {
+        id: "review-message".into(),
+        msg: EventMsg::AgentMessage(AgentMessageEvent {
+            message: "Review progress update".to_string(),
+            phase: None,
+        }),
+    });
+
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("Review progress update"));
+}
+
+#[tokio::test]
+async fn thread_snapshot_replay_preserves_agent_message_during_review_mode() {
+    let (mut chat, mut rx, _ops) = make_chatwidget_manual(None).await;
+
+    chat.handle_codex_event_replay(Event {
+        id: "review-start".into(),
+        msg: EventMsg::EnteredReviewMode(ReviewRequest {
+            target: ReviewTarget::UncommittedChanges,
+            user_facing_hint: None,
+        }),
+    });
+    let _ = drain_insert_history(&mut rx);
+
+    chat.handle_codex_event_replay(Event {
+        id: "review-message".into(),
+        msg: EventMsg::AgentMessage(AgentMessageEvent {
+            message: "Review progress update".to_string(),
+            phase: None,
+        }),
+    });
+
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("Review progress update"));
+}
+
 /// Exiting review restores the pre-review context window indicator.
 #[tokio::test]
 async fn review_restores_context_window_indicator() {
@@ -1720,6 +1812,7 @@ async fn make_chatwidget_manual(
         show_welcome_banner: true,
         startup_tooltip_override: None,
         queued_user_messages: VecDeque::new(),
+        pending_steers: VecDeque::new(),
         queued_message_edit_binding: crate::key_hint::alt(KeyCode::Up),
         suppress_session_configured_redraw: false,
         pending_notification: None,
@@ -2753,6 +2846,94 @@ async fn plan_implementation_popup_shows_after_proposed_plan_output() {
     );
 }
 
+#[tokio::test]
+async fn plan_implementation_popup_skips_when_steer_follows_proposed_plan() {
+    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5")).await;
+    chat.set_feature_enabled(Feature::CollaborationModes, true);
+    let plan_mask =
+        collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan)
+            .expect("expected plan collaboration mask");
+    chat.set_collaboration_mask(plan_mask);
+    chat.thread_id = Some(ThreadId::new());
+
+    chat.on_task_started();
+    chat.on_plan_item_completed(
+        "- Step 1
+- Step 2
+"
+        .to_string(),
+    );
+    chat.bottom_pane
+        .set_composer_text("Please continue.".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "Please continue.".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+
+    complete_user_message(&mut chat, "user-1", "Please continue.");
+    chat.on_task_complete(None, false);
+
+    let popup = render_bottom_popup(&chat, 80);
+    assert!(
+        !popup.contains(PLAN_IMPLEMENTATION_TITLE),
+        "expected no plan popup after a steer follows the plan, got {popup:?}"
+    );
+}
+
+#[tokio::test]
+async fn plan_implementation_popup_shows_after_new_plan_follows_steer() {
+    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5")).await;
+    chat.set_feature_enabled(Feature::CollaborationModes, true);
+    let plan_mask =
+        collaboration_modes::mask_for_kind(chat.models_manager.as_ref(), ModeKind::Plan)
+            .expect("expected plan collaboration mask");
+    chat.set_collaboration_mask(plan_mask);
+    chat.thread_id = Some(ThreadId::new());
+
+    chat.on_task_started();
+    chat.on_plan_item_completed(
+        "- Initial plan
+"
+        .to_string(),
+    );
+    chat.bottom_pane
+        .set_composer_text("Please revise.".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "Please revise.".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+
+    complete_user_message(&mut chat, "user-1", "Please revise.");
+    chat.on_plan_item_completed(
+        "- Revised plan
+"
+        .to_string(),
+    );
+    chat.on_task_complete(None, false);
+
+    let popup = render_bottom_popup(&chat, 80);
+    assert!(
+        popup.contains(PLAN_IMPLEMENTATION_TITLE),
+        "expected plan popup after a newer plan follows the steer, got {popup:?}"
+    );
+}
+
 #[tokio::test]
 async fn plan_implementation_popup_skips_when_rate_limit_prompt_pending() {
     let (mut chat, _rx, _op_rx) = make_chatwidget_manual(Some("gpt-5")).await;
@@ -3064,6 +3245,41 @@ fn complete_assistant_message(
     });
 }
 
+fn pending_steer(text: &str) -> PendingSteer {
+    PendingSteer {
+        user_message: UserMessage::from(text),
+        compare_key: PendingSteerCompareKey {
+            message: text.to_string(),
+            image_count: 0,
+        },
+    }
+}
+
+fn complete_user_message(chat: &mut ChatWidget, item_id: &str, text: &str) {
+    complete_user_message_for_inputs(
+        chat,
+        item_id,
+        vec![UserInput::Text {
+            text: text.to_string(),
+            text_elements: Vec::new(),
+        }],
+    );
+}
+
+fn complete_user_message_for_inputs(chat: &mut ChatWidget, item_id: &str, content: Vec<UserInput>) {
+    chat.handle_codex_event(Event {
+        id: format!("raw-{item_id}"),
+        msg: EventMsg::ItemCompleted(ItemCompletedEvent {
+            thread_id: ThreadId::new(),
+            turn_id: "turn-1".to_string(),
+            item: TurnItem::UserMessage(UserMessageItem {
+                id: item_id.to_string(),
+                content,
+            }),
+        }),
+    });
+}
+
 fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) -> ExecCommandBeginEvent {
     begin_exec_with_source(chat, call_id, raw_cmd, ExecCommandSource::Agent)
 }
@@ -3166,7 +3382,7 @@ async fn alt_up_edits_most_recent_queued_message() {
         .push_back(UserMessage::from("first queued".to_string()));
     chat.queued_user_messages
         .push_back(UserMessage::from("second queued".to_string()));
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     // Press Alt+Up to edit the most recent (last) queued message.
     chat.handle_key_event(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT));
@@ -3200,7 +3416,7 @@ async fn assert_shift_left_edits_most_recent_queued_message_for_terminal(
         .push_back(UserMessage::from("first queued".to_string()));
     chat.queued_user_messages
         .push_back(UserMessage::from("second queued".to_string()));
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     // Press Shift+Left to edit the most recent (last) queued message.
     chat.handle_key_event(KeyEvent::new(KeyCode::Left, KeyModifiers::SHIFT));
@@ -3459,8 +3675,8 @@ async fn unified_exec_begin_restores_working_status_snapshot() {
 }
 
 #[tokio::test]
-async fn enter_queues_while_plan_stream_is_active() {
-    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;
+async fn steer_enter_queues_while_plan_stream_is_active() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
     chat.thread_id = Some(ThreadId::new());
     chat.set_feature_enabled(Feature::CollaborationModes, true);
     let plan_mask =
@@ -3469,6 +3685,7 @@ async fn enter_queues_while_plan_stream_is_active() {
     chat.set_collaboration_mask(plan_mask);
     chat.on_task_started();
     chat.on_plan_delta("- Step 1".to_string());
+    let _ = drain_insert_history(&mut rx);
 
     chat.bottom_pane
         .set_composer_text("queued submission".to_string(), Vec::new(), Vec::new());
@@ -3480,12 +3697,44 @@ async fn enter_queues_while_plan_stream_is_active() {
         chat.queued_user_messages.front().unwrap().text,
         "queued submission"
     );
-    assert_matches!(op_rx.try_recv(), Err(TryRecvError::Empty));
+    assert!(chat.pending_steers.is_empty());
+    assert_no_submit_op(&mut op_rx);
+    assert!(drain_insert_history(&mut rx).is_empty());
 }
 
 #[tokio::test]
-async fn steer_enter_queues_while_final_answer_stream_is_active() {
-    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;
+async fn steer_enter_uses_pending_steers_while_turn_is_running_without_streaming() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+
+    chat.bottom_pane
+        .set_composer_text("queued while running".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    assert!(chat.queued_user_messages.is_empty());
+    assert_eq!(chat.pending_steers.len(), 1);
+    assert_eq!(
+        chat.pending_steers.front().unwrap().user_message.text,
+        "queued while running"
+    );
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { .. } => {}
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    complete_user_message(&mut chat, "user-1", "queued while running");
+
+    assert!(chat.pending_steers.is_empty());
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("queued while running"));
+}
+
+#[tokio::test]
+async fn steer_enter_uses_pending_steers_while_final_answer_stream_is_active() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
     chat.thread_id = Some(ThreadId::new());
     chat.on_task_started();
     // Keep the assistant stream open (no commit tick/finalize) to model the repro window:
@@ -3499,26 +3748,228 @@ async fn steer_enter_queues_while_final_answer_stream_is_active() {
     );
     chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
 
-    assert_eq!(chat.queued_user_messages.len(), 1);
+    assert!(chat.queued_user_messages.is_empty());
+    assert_eq!(chat.pending_steers.len(), 1);
     assert_eq!(
-        chat.queued_user_messages.front().unwrap().text,
+        chat.pending_steers.front().unwrap().user_message.text,
         "queued while streaming"
     );
-    assert_no_submit_op(&mut op_rx);
-
-    // Once final output ends, the queued input must be submitted automatically.
-    chat.on_task_complete(None, false);
-
-    assert!(chat.queued_user_messages.is_empty());
     match next_submit_op(&mut op_rx) {
         Op::UserTurn { .. } => {}
-        other => panic!("expected Op::UserTurn after stream completion, got {other:?}"),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
     }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    complete_user_message(&mut chat, "user-1", "queued while streaming");
+
+    assert!(chat.pending_steers.is_empty());
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("queued while streaming"));
+}
+
+#[tokio::test]
+async fn failed_pending_steer_submit_does_not_add_pending_preview() {
+    let (mut chat, mut rx, op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+    drop(op_rx);
+
+    chat.bottom_pane.set_composer_text(
+        "queued while streaming".to_string(),
+        Vec::new(),
+        Vec::new(),
+    );
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    assert!(chat.pending_steers.is_empty());
+    assert!(chat.queued_user_messages.is_empty());
+    assert!(drain_insert_history(&mut rx).is_empty());
+}
+
+#[tokio::test]
+async fn live_legacy_agent_message_after_item_completed_does_not_duplicate_assistant_message() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+
+    complete_assistant_message(
+        &mut chat,
+        "msg-live",
+        "hello",
+        Some(MessagePhase::FinalAnswer),
+    );
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("hello"));
+
+    chat.handle_codex_event(Event {
+        id: "legacy-live".into(),
+        msg: EventMsg::AgentMessage(AgentMessageEvent {
+            message: "hello".into(),
+            phase: Some(MessagePhase::FinalAnswer),
+        }),
+    });
+
+    assert!(drain_insert_history(&mut rx).is_empty());
+}
+
+#[test]
+fn rendered_user_message_event_from_inputs_matches_flattened_user_message_shape() {
+    let local_image = PathBuf::from("/tmp/local.png");
+    let rendered = ChatWidget::rendered_user_message_event_from_inputs(&[
+        UserInput::Text {
+            text: "hello ".to_string(),
+            text_elements: vec![TextElement::new((0..5).into(), None)],
+        },
+        UserInput::Image {
+            image_url: "https://example.com/remote.png".to_string(),
+        },
+        UserInput::LocalImage {
+            path: local_image.clone(),
+        },
+        UserInput::Skill {
+            name: "demo".to_string(),
+            path: PathBuf::from("/tmp/skill/SKILL.md"),
+        },
+        UserInput::Mention {
+            name: "repo".to_string(),
+            path: "app://repo".to_string(),
+        },
+        UserInput::Text {
+            text: "world".to_string(),
+            text_elements: vec![TextElement::new((0..5).into(), Some("planet".to_string()))],
+        },
+    ]);
+
+    assert_eq!(
+        rendered,
+        ChatWidget::rendered_user_message_event_from_parts(
+            "hello world".to_string(),
+            vec![
+                TextElement::new((0..5).into(), Some("hello".to_string())),
+                TextElement::new((6..11).into(), Some("planet".to_string())),
+            ],
+            vec![local_image],
+            vec!["https://example.com/remote.png".to_string()],
+        )
+    );
+}
+
+#[tokio::test]
+async fn item_completed_only_pops_front_pending_steer() {
+    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+    chat.pending_steers.push_back(pending_steer("first"));
+    chat.pending_steers.push_back(pending_steer("second"));
+    chat.refresh_pending_input_preview();
+
+    complete_user_message(&mut chat, "user-other", "other");
+
+    assert_eq!(chat.pending_steers.len(), 2);
+    assert_eq!(
+        chat.pending_steers.front().unwrap().user_message.text,
+        "first"
+    );
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("other"));
+
+    complete_user_message(&mut chat, "user-first", "first");
+
+    assert_eq!(chat.pending_steers.len(), 1);
+    assert_eq!(
+        chat.pending_steers.front().unwrap().user_message.text,
+        "second"
+    );
+    let inserted = drain_insert_history(&mut rx);
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("first"));
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn item_completed_pops_pending_steer_with_local_image_and_text_elements() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+
+    let temp = tempdir().expect("tempdir");
+    let image_path = temp.path().join("pending-steer.png");
+    const TINY_PNG_BYTES: &[u8] = &[
+        137, 80, 78, 71, 13, 10, 26, 10, 0, 0, 0, 13, 73, 72, 68, 82, 0, 0, 0, 1, 0, 0, 0, 1, 8, 6,
+        0, 0, 0, 31, 21, 196, 137, 0, 0, 0, 11, 73, 68, 65, 84, 120, 156, 99, 96, 0, 2, 0, 0, 5, 0,
+        1, 122, 94, 171, 63, 0, 0, 0, 0, 73, 69, 78, 68, 174, 66, 96, 130,
+    ];
+    std::fs::write(&image_path, TINY_PNG_BYTES).expect("write image");
+
+    let text = "note".to_string();
+    let text_elements = vec![TextElement::new((0..4).into(), Some("note".to_string()))];
+    chat.submit_user_message(UserMessage {
+        text: text.clone(),
+        local_images: vec![LocalImageAttachment {
+            placeholder: "[Image #1]".to_string(),
+            path: image_path,
+        }],
+        remote_image_urls: Vec::new(),
+        text_elements,
+        mention_bindings: Vec::new(),
+    });
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { .. } => {}
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+
+    assert_eq!(chat.pending_steers.len(), 1);
+    let pending = chat.pending_steers.front().unwrap();
+    assert_eq!(pending.user_message.local_images.len(), 1);
+    assert_eq!(pending.user_message.text_elements.len(), 1);
+    assert_eq!(pending.compare_key.message, text);
+    assert_eq!(pending.compare_key.image_count, 1);
+
+    complete_user_message_for_inputs(
+        &mut chat,
+        "user-1",
+        vec![
+            UserInput::Image {
+                image_url: "data:image/png;base64,placeholder".to_string(),
+            },
+            UserInput::Text {
+                text,
+                text_elements: Vec::new(),
+            },
+        ],
+    );
+
+    assert!(chat.pending_steers.is_empty());
+
+    let mut user_cell = None;
+    while let Ok(ev) = rx.try_recv() {
+        if let AppEvent::InsertHistoryCell(cell) = ev
+            && let Some(cell) = cell.as_any().downcast_ref::<UserHistoryCell>()
+        {
+            user_cell = Some((
+                cell.message.clone(),
+                cell.text_elements.clone(),
+                cell.local_image_paths.clone(),
+                cell.remote_image_urls.clone(),
+            ));
+            break;
+        }
+    }
+
+    let (stored_message, stored_elements, stored_images, stored_remote_image_urls) =
+        user_cell.expect("expected pending steer user history cell");
+    assert_eq!(stored_message, "note");
+    assert_eq!(
+        stored_elements,
+        vec![TextElement::new((0..4).into(), Some("note".to_string()))]
+    );
+    assert_eq!(stored_images.len(), 1);
+    assert!(stored_images[0].ends_with("pending-steer.png"));
+    assert!(stored_remote_image_urls.is_empty());
 }
 
 #[tokio::test]
 async fn steer_enter_during_final_stream_preserves_follow_up_prompts_in_order() {
-    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
     chat.thread_id = Some(ThreadId::new());
     chat.on_task_started();
     // Simulate "dead mode" repro timing by keeping a final-answer stream active while the
@@ -3532,19 +3983,16 @@ async fn steer_enter_during_final_stream_preserves_follow_up_prompts_in_order()
         .set_composer_text("second follow-up".to_string(), Vec::new(), Vec::new());
     chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
 
-    assert_eq!(chat.queued_user_messages.len(), 2);
+    assert!(chat.queued_user_messages.is_empty());
+    assert_eq!(chat.pending_steers.len(), 2);
     assert_eq!(
-        chat.queued_user_messages.front().unwrap().text,
+        chat.pending_steers.front().unwrap().user_message.text,
         "first follow-up"
     );
     assert_eq!(
-        chat.queued_user_messages.back().unwrap().text,
+        chat.pending_steers.back().unwrap().user_message.text,
         "second follow-up"
     );
-    assert_no_submit_op(&mut op_rx);
-
-    // Completion must recover by submitting the oldest queued prompt first.
-    chat.on_task_complete(None, false);
 
     let first_items = match next_submit_op(&mut op_rx) {
         Op::UserTurn { items, .. } => items,
@@ -3557,17 +4005,6 @@ async fn steer_enter_during_final_stream_preserves_follow_up_prompts_in_order()
             text_elements: Vec::new(),
         }]
     );
-    assert_eq!(chat.queued_user_messages.len(), 1);
-    assert_eq!(
-        chat.queued_user_messages.front().unwrap().text,
-        "second follow-up"
-    );
-
-    // A subsequent turn lifecycle should continue draining remaining queued prompts, proving
-    // the widget did not enter a permanently stuck state.
-    chat.on_task_started();
-    chat.on_task_complete(None, false);
-
     let second_items = match next_submit_op(&mut op_rx) {
         Op::UserTurn { items, .. } => items,
         other => panic!("expected Op::UserTurn, got {other:?}"),
@@ -3579,7 +4016,208 @@ async fn steer_enter_during_final_stream_preserves_follow_up_prompts_in_order()
             text_elements: Vec::new(),
         }]
     );
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    complete_user_message(&mut chat, "user-1", "first follow-up");
+
+    assert_eq!(chat.pending_steers.len(), 1);
+    assert_eq!(
+        chat.pending_steers.front().unwrap().user_message.text,
+        "second follow-up"
+    );
+    let first_insert = drain_insert_history(&mut rx);
+    assert_eq!(first_insert.len(), 1);
+    assert!(lines_to_single_string(&first_insert[0]).contains("first follow-up"));
+
+    complete_user_message(&mut chat, "user-2", "second follow-up");
+
+    assert!(chat.pending_steers.is_empty());
+    let second_insert = drain_insert_history(&mut rx);
+    assert_eq!(second_insert.len(), 1);
+    assert!(lines_to_single_string(&second_insert[0]).contains("second follow-up"));
+}
+
+#[tokio::test]
+async fn manual_interrupt_restores_pending_steers_to_composer() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+    chat.on_agent_message_delta(
+        "Final answer line
+"
+        .to_string(),
+    );
+
+    chat.bottom_pane.set_composer_text(
+        "queued while streaming".to_string(),
+        Vec::new(),
+        Vec::new(),
+    );
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    assert_eq!(chat.pending_steers.len(), 1);
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "queued while streaming".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    chat.on_interrupted_turn(TurnAbortReason::Interrupted);
+
+    assert!(chat.pending_steers.is_empty());
+    assert_eq!(chat.bottom_pane.composer_text(), "queued while streaming");
+    assert_no_submit_op(&mut op_rx);
+
+    let inserted = drain_insert_history(&mut rx);
+    assert!(
+        inserted
+            .iter()
+            .all(|cell| !lines_to_single_string(cell).contains("queued while streaming"))
+    );
+}
+
+#[tokio::test]
+async fn manual_interrupt_restores_pending_steer_mention_bindings_to_composer() {
+    let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+    chat.on_agent_message_delta("Final answer line\n".to_string());
+
+    let mention_bindings = vec![MentionBinding {
+        mention: "figma".to_string(),
+        path: "/tmp/skills/figma/SKILL.md".to_string(),
+    }];
+    chat.bottom_pane.set_composer_text_with_mention_bindings(
+        "please use $figma".to_string(),
+        vec![TextElement::new(
+            (11..17).into(),
+            Some("$figma".to_string()),
+        )],
+        Vec::new(),
+        mention_bindings.clone(),
+    );
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "please use $figma".to_string(),
+                text_elements: vec![TextElement::new(
+                    (11..17).into(),
+                    Some("$figma".to_string()),
+                )],
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+
+    chat.on_interrupted_turn(TurnAbortReason::Interrupted);
+
+    assert_eq!(chat.bottom_pane.composer_text(), "please use $figma");
+    assert_eq!(chat.bottom_pane.take_mention_bindings(), mention_bindings);
+    assert_no_submit_op(&mut op_rx);
+}
+
+#[tokio::test]
+async fn manual_interrupt_restores_pending_steers_before_queued_messages() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+    chat.on_agent_message_delta(
+        "Final answer line
+"
+        .to_string(),
+    );
+
+    chat.bottom_pane
+        .set_composer_text("pending steer".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+    chat.queued_user_messages
+        .push_back(UserMessage::from("queued draft".to_string()));
+    chat.refresh_pending_input_preview();
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "pending steer".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    chat.on_interrupted_turn(TurnAbortReason::Interrupted);
+
+    assert!(chat.pending_steers.is_empty());
     assert!(chat.queued_user_messages.is_empty());
+    assert_eq!(
+        chat.bottom_pane.composer_text(),
+        "pending steer
+queued draft"
+    );
+    assert_no_submit_op(&mut op_rx);
+}
+
+#[tokio::test]
+async fn replaced_turn_clears_pending_steers_but_keeps_queued_drafts() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+    chat.on_agent_message_delta(
+        "Final answer line
+"
+        .to_string(),
+    );
+
+    chat.bottom_pane
+        .set_composer_text("pending steer".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+    chat.queued_user_messages
+        .push_back(UserMessage::from("queued draft".to_string()));
+    chat.refresh_pending_input_preview();
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "pending steer".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    chat.handle_codex_event(Event {
+        id: "replaced".into(),
+        msg: EventMsg::TurnAborted(codex_protocol::protocol::TurnAbortedEvent {
+            turn_id: Some("turn-1".to_string()),
+            reason: TurnAbortReason::Replaced,
+        }),
+    });
+
+    assert!(chat.pending_steers.is_empty());
+    assert!(chat.queued_user_messages.is_empty());
+    assert_eq!(chat.bottom_pane.composer_text(), "");
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { items, .. } => assert_eq!(
+            items,
+            vec![UserInput::Text {
+                text: "queued draft".to_string(),
+                text_elements: Vec::new(),
+            }]
+        ),
+        other => panic!("expected queued draft Op::UserTurn, got {other:?}"),
+    }
 }
 
 #[tokio::test]
@@ -3998,13 +4636,7 @@ async fn unified_exec_wait_after_final_agent_message_snapshot() {
     begin_unified_exec_startup(&mut chat, "call-wait", "proc-1", "cargo test -p codex-core");
     terminal_interaction(&mut chat, "call-wait-stdin", "proc-1", "");
 
-    chat.handle_codex_event(Event {
-        id: "turn-1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message: "Final response.".into(),
-            phase: None,
-        }),
-    });
+    complete_assistant_message(&mut chat, "msg-1", "Final response.", None);
     chat.handle_codex_event(Event {
         id: "turn-1".into(),
         msg: EventMsg::TurnComplete(TurnCompleteEvent {
@@ -4805,7 +5437,7 @@ async fn slash_copy_state_clears_on_thread_rollback() {
 async fn slash_copy_is_unavailable_when_legacy_agent_message_is_not_repeated_on_turn_complete() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
 
-    chat.handle_codex_event(Event {
+    chat.handle_codex_event_replay(Event {
         id: "turn-1".into(),
         msg: EventMsg::AgentMessage(AgentMessageEvent {
             message: "Legacy final message".into(),
@@ -7110,7 +7742,7 @@ async fn interrupt_restores_queued_messages_into_composer() {
         .push_back(UserMessage::from("first queued".to_string()));
     chat.queued_user_messages
         .push_back(UserMessage::from("second queued".to_string()));
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     // Deliver a TurnAborted event with Interrupted reason (as if Esc was pressed).
     chat.handle_codex_event(Event {
@@ -7150,7 +7782,7 @@ async fn interrupt_prepends_queued_messages_before_existing_composer_text() {
         .push_back(UserMessage::from("first queued".to_string()));
     chat.queued_user_messages
         .push_back(UserMessage::from("second queued".to_string()));
-    chat.refresh_queued_user_messages();
+    chat.refresh_pending_input_preview();
 
     chat.handle_codex_event(Event {
         id: "turn-1".into(),
@@ -8395,22 +9027,10 @@ async fn multiple_agent_messages_in_single_turn_emit_multiple_headers() {
     });
 
     // First finalized assistant message
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message: "First message".into(),
-            phase: None,
-        }),
-    });
+    complete_assistant_message(&mut chat, "msg-first", "First message", None);
 
     // Second finalized assistant message in the same turn
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message: "Second message".into(),
-            phase: None,
-        }),
-    });
+    complete_assistant_message(&mut chat, "msg-second", "Second message", None);
 
     // End turn
     chat.handle_codex_event(Event {
@@ -8450,13 +9070,7 @@ async fn final_reasoning_then_message_without_deltas_are_rendered() {
             text: "I will first analyze the request.".into(),
         }),
     });
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message: "Here is the result.".into(),
-            phase: None,
-        }),
-    });
+    complete_assistant_message(&mut chat, "msg-result", "Here is the result.", None);
 
     // Drain history and snapshot the combined visible content.
     let cells = drain_insert_history(&mut rx);
@@ -8467,83 +9081,18 @@ async fn final_reasoning_then_message_without_deltas_are_rendered() {
     assert_snapshot!(combined);
 }
 
-#[tokio::test]
-async fn deltas_then_same_final_message_are_rendered_snapshot() {
-    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
-
-    // Stream some reasoning deltas first.
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent {
-            delta: "I will ".into(),
-        }),
-    });
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent {
-            delta: "first analyze the ".into(),
-        }),
-    });
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentReasoningDelta(AgentReasoningDeltaEvent {
-            delta: "request.".into(),
-        }),
-    });
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentReasoning(AgentReasoningEvent {
-            text: "request.".into(),
-        }),
-    });
-
-    // Then stream answer deltas, followed by the exact same final message.
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
-            delta: "Here is the ".into(),
-        }),
-    });
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessageDelta(AgentMessageDeltaEvent {
-            delta: "result.".into(),
-        }),
-    });
-
-    chat.handle_codex_event(Event {
-        id: "s1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message: "Here is the result.".into(),
-            phase: None,
-        }),
-    });
-
-    // Snapshot the combined visible content to ensure we render as expected
-    // when deltas are followed by the identical final message.
-    let cells = drain_insert_history(&mut rx);
-    let combined = cells
-        .iter()
-        .map(|lines| lines_to_single_string(lines))
-        .collect::<String>();
-    assert_snapshot!(combined);
-}
-
 // Combined visual snapshot using vt100 for history + direct buffer overlay for UI.
 // This renders the final visual as seen in a terminal: history above, then a blank line,
 // then the exec block, another blank line, the status line, a blank line, and the composer.
 #[tokio::test]
 async fn chatwidget_exec_and_status_layout_vt100_snapshot() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
-    chat.handle_codex_event(Event {
-        id: "t1".into(),
-        msg: EventMsg::AgentMessage(AgentMessageEvent {
-            message:
-                "I’m going to search the repo for where “Change Approved” is rendered to update that view."
-                    .into(),
-            phase: None,
-        }),
-    });
+    complete_assistant_message(
+        &mut chat,
+        "msg-search",
+        "I’m going to search the repo for where “Change Approved” is rendered to update that view.",
+        None,
+    );
 
     let command = vec!["bash".into(), "-lc".into(), "rg \"Change Approved\"".into()];
     let parsed_cmd = vec![
@@ -8757,6 +9306,37 @@ async fn chatwidget_tall() {
     assert_snapshot!(term.backend().vt100().screen().contents());
 }
 
+#[tokio::test]
+async fn enter_queues_user_messages_while_review_is_running() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+
+    chat.handle_codex_event(Event {
+        id: "review-1".into(),
+        msg: EventMsg::EnteredReviewMode(ReviewRequest {
+            target: ReviewTarget::UncommittedChanges,
+            user_facing_hint: Some("current changes".to_string()),
+        }),
+    });
+    let _ = drain_insert_history(&mut rx);
+
+    chat.bottom_pane.set_composer_text(
+        "Queued while /review is running.".to_string(),
+        Vec::new(),
+        Vec::new(),
+    );
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    assert_eq!(chat.queued_user_messages.len(), 1);
+    assert_eq!(
+        chat.queued_user_messages.front().unwrap().text,
+        "Queued while /review is running."
+    );
+    assert!(chat.pending_steers.is_empty());
+    assert_no_submit_op(&mut op_rx);
+    assert!(drain_insert_history(&mut rx).is_empty());
+}
+
 #[tokio::test]
 async fn review_queues_user_messages_snapshot() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;