From 4a03589330586a61690b753c4e07fb3d8fc07752 Mon Sep 17 00:00:00 2001 From: Charles Cunningham Date: Mon, 2 Mar 2026 15:13:14 -0800 Subject: [PATCH] core: emit leftover user-message lifecycle --- codex-rs/core/src/codex.rs | 72 +++++++++++++++++++++++++++- codex-rs/core/src/tasks/mod.rs | 20 +++++++- codex-rs/tui/src/chatwidget.rs | 15 ------ codex-rs/tui/src/chatwidget/tests.rs | 24 ---------- 4 files changed, 88 insertions(+), 43 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 202eace800..ca2c329614 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -6554,6 +6554,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; @@ -9265,8 +9266,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(), @@ -9281,6 +9282,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 { @@ -9306,6 +9309,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 97719f1049..8d95b1c9e5 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,22 @@ impl Session { .into_iter() .map(ResponseItem::from) .collect::>(); - 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) { + 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/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 2baee03080..bc95ec4a4c 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -1510,21 +1510,6 @@ impl ChatWidget { self.request_redraw(); let had_pending_steers = !self.pending_steers.is_empty(); - while let Some(pending_steer) = self.pending_steers.pop_front() { - self.last_rendered_user_message_event = Some(pending_steer.clone()); - if !pending_steer.message.trim().is_empty() - || !pending_steer.text_elements.is_empty() - || !pending_steer.remote_image_urls.is_empty() - { - self.add_to_history(history_cell::new_user_prompt( - pending_steer.message, - pending_steer.text_elements, - pending_steer.local_images, - pending_steer.remote_image_urls, - )); - } - self.needs_final_message_separator = false; - } self.refresh_pending_input_preview(); if !from_replay && self.queued_user_messages.is_empty() && !had_pending_steers { diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index b57d1d8e2a..00764059fd 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -3973,30 +3973,6 @@ queued draft" assert_no_submit_op(&mut op_rx); } -#[tokio::test] -async fn task_complete_renders_unacknowledged_pending_steer() { - 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("late steer".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 { .. } => {} - other => panic!("expected Op::UserTurn, got {other:?}"), - } - assert!(drain_insert_history(&mut rx).is_empty()); - - chat.on_task_complete(None, false); - - 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("late steer")); -} - #[tokio::test] async fn enter_submits_when_plan_stream_is_not_active() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(None).await;