openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-13 21:30:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve thread input across TUI session resumes (#61216)

Browse Source 
## Why

Resuming another session must keep drafts, queued follow-ups, and interrupted steer retries with their original thread. Submitted input whose delivery is still uncertain must wait for matching receipts before the queue can advance.

## What changed

- Pause submissions while draining outgoing events, then capture input before switching threads or backends.
- Restore saved input through a shared resume path, reconcile receipts from resumed history and buffered events, preserve active-turn state, and dispatch eligible queued input when the resumed thread is idle.
- Preserve whether a draft's leading `!` is literal text or a shell command.
- Reject `!` shell commands in Cloud conversations with an error and clear the pending submission so the next prompt can proceed.

## Testing

Add regression coverage for session round trips with buffered completion and interruption events, receipt reconciliation across read-only and writable resumes, draft shell mode, and submitting a prompt after a rejected Cloud shell command.

GitOrigin-RevId: 9a823a398a371e689b8ae743de494323ae95ef57
This commit is contained in:
Benjamin Carlsson
2026-11-13 00:52:53 +00:00
committed by copyberry
parent 006911a313
commit 5183e1dcb2
20 changed files with 841 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/agents_overview.rs
+2
View File
@@ -1055,6 +1055,8 @@ impl App {
&& self.active_turn_id_for_thread(thread_id).await.is_some();
self.reconcile_saved_session_input(thread_id, &mut input)
.await;
// Missing receipts must not turn a previously submitted steer into unsent input.
input.reconnect_pending |= !preserve_in_flight_turn && !input.pending_steers.is_empty();
self.chat_widget.restore_thread_input_state(
Some(input),
ThreadInputStateRestoreMode {
codex-rs/tui/src/app/agents_overview_tests.rs
+277 -17
View File
@@ -2581,13 +2581,99 @@ async fn resume_failure_keeps_command_center_available() {
}
#[tokio::test]
async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
let mut app = Box::pin(make_test_app()).await;
async fn resume_round_trip_preserves_input_and_dispatches_idle_queue() -> Result<()> {
check_resume_round_trip(ResumeRoundTrip::CompleteWhileAway).await
}
#[tokio::test]
async fn resume_round_trip_drains_buffered_completion_without_sending_on_other_thread() -> Result<()>
{
check_resume_round_trip(ResumeRoundTrip::BufferCompletion).await
}
#[tokio::test]
async fn resume_round_trip_keeps_buffered_esc_retry_on_original_thread() -> Result<()> {
check_resume_round_trip(ResumeRoundTrip::BufferInterruption).await
}
#[derive(PartialEq, Eq)]
enum ResumeRoundTrip {
CompleteWhileAway,
BufferCompletion,
BufferInterruption,
}
async fn check_resume_round_trip(round_trip: ResumeRoundTrip) -> Result<()> {
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_model_provider_info::ModelProviderInfo;
use core_test_support::responses;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
async fn wait_for_turn_status(
server: &mut AppServerSession,
outgoing: ThreadId,
status: TurnStatus,
) -> Result<codex_app_server_protocol::Turn> {
tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async {
loop {
let thread = server.thread_read(outgoing, /*include_turns*/ true).await?;
if let Some(turn) = thread.turns.into_iter().last()
&& turn.status == status
{
break Ok::<_, color_eyre::eyre::Report>(turn);
}
tokio::task::yield_now().await;
}
})
.await?
}
let interrupted = round_trip == ResumeRoundTrip::BufferInterruption;
let requests_before_return = if interrupted { 1 } else { 2 };
let steer_text = "Steer for the first task";
let (release, gate) = tokio::sync::oneshot::channel();
let (responses_server, _completions) = start_streaming_sse_server(vec![
vec![
StreamingSseChunk {
gate: None,
body: responses::sse(vec![responses::ev_response_created("running")]),
},
StreamingSseChunk {
gate: Some(gate),
body: responses::sse(vec![responses::ev_completed("running")]),
},
],
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![responses::ev_completed("steer")]),
}],
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![responses::ev_completed("follow-up")]),
}],
])
.await;
let (mut app, mut events, _ops) = Box::pin(make_test_app_with_channels()).await;
trust_fixture_folders(&mut app);
std::fs::write(
app.config.codex_home.join("config.toml"),
"[tui]\nresume_cwd = \"current\"\n",
)?;
app.config.model = Some("gpt-5.2".into());
app.config.model_provider_id = "resume-test".into();
app.config.model_provider = ModelProviderInfo {
name: "Resume test".into(),
base_url: Some(format!("{}/v1", responses_server.uri())),
request_max_retries: Some(0),
stream_max_retries: Some(0),
..ModelProviderInfo::default()
};
app_test_support::MockResponsesConfig::new(responses_server.uri())
.with_model("gpt-5.2")
.with_model_provider("resume-test")
.with_provider_name("Resume test")
.with_extra_config("[tui]\nresume_cwd = \"current\"")
.write(app.config.codex_home.as_path())?;
let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?;
let backend_id = server.backend_id();
let mut targets = Vec::new();
@@ -2618,7 +2704,68 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
}
let mut tui = crate::tui::test_support::make_test_tui()?;
let mut expected_states = Vec::new();
let mut release = Some(release);
for target in targets.iter().chain(&targets) {
if round_trip != ResumeRoundTrip::CompleteWhileAway
&& expected_states.len() == targets.len()
&& target.thread_id == targets[1].thread_id
{
let outgoing = targets[0].thread_id;
assert_eq!(app.chat_widget.thread_id(), Some(outgoing));
assert_eq!(
app.chat_widget
.capture_thread_input_state()
.unwrap()
.pending_steers
.len(),
1
);
tokio::time::timeout(
Duration::from_secs(/*secs*/ 5),
responses_server.wait_for_request_count(requests_before_return),
)
.await?;
assert_eq!(
responses_server.requests().await.len(),
requests_before_return
);
let status = if interrupted {
let turn_id = app
.chat_widget
.last_turn_id()
.expect("running turn")
.to_owned();
app.chat_widget
.handle_key_event(KeyEvent::from(KeyCode::Esc));
assert!(
std::iter::from_fn(|| events.try_recv().ok())
.any(|event| matches!(event, AppEvent::CodexOp(AppCommand::Interrupt)))
);
server.turn_interrupt(outgoing, turn_id).await?;
TurnStatus::Interrupted
} else {
release.take().unwrap().send(()).unwrap();
TurnStatus::Completed
};
let turn = wait_for_turn_status(&mut server, outgoing, status).await?;
drop(release.take());
// Route the real completion to A's receiver, leaving it buffered across the switch.
app.enqueue_thread_notification(
outgoing,
ServerNotification::TurnCompleted(
codex_app_server_protocol::TurnCompletedNotification {
thread_id: outgoing.to_string(),
turn,
},
),
)
.await?;
assert!(
app.active_thread_rx
.as_ref()
.is_some_and(|rx| !rx.is_empty())
);
}
let view = app.agents_overview_view(Vec::new(), /*selected_thread_id*/ None);
app.chat_widget.show_bottom_pane_view(Box::new(view));
Box::pin(app.apply_resume_picker_selection(
@@ -2631,11 +2778,35 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
if expected_states.len() < targets.len() {
assert_eq!(app.chat_widget.composer_text_with_pending(), "");
assert!(app.chat_widget.queued_user_message_texts().is_empty());
let running_turn = if target.thread_id == targets[0].thread_id {
let response = server
.request_handle()
.request_typed::<TurnStartResponse>(ClientRequest::TurnStart {
request_id: RequestId::String(Uuid::new_v4().to_string()),
params: TurnStartParams {
thread_id: target.thread_id.to_string(),
input: vec![codex_app_server_protocol::UserInput::Text {
text: "Keep working".into(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?;
tokio::time::timeout(
Duration::from_secs(/*secs*/ 5),
responses_server.wait_for_request_count(/*count*/ 1),
)
.await?;
Some(response.turn)
} else {
None
};
app.chat_widget.handle_server_notification(
ServerNotification::TurnStarted(
codex_app_server_protocol::TurnStartedNotification {
thread_id: target.thread_id.to_string(),
turn: codex_app_server_protocol::Turn {
turn: running_turn.unwrap_or_else(|| codex_app_server_protocol::Turn {
id: "turn-with-follow-up".to_string(),
root_turn_id: None,
items_view: codex_app_server_protocol::TurnItemsView::Full,
@@ -2645,7 +2816,7 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
started_at: None,
completed_at: None,
duration_ms: None,
},
}),
},
),
/*replay_kind*/ None,
@@ -2657,15 +2828,17 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
assert_eq!(app.chat_widget.queued_user_message_texts(), vec![follow_up]);
app.chat_widget
.apply_external_edit(format!("Draft for {}", target.thread_id));
let mut input_state = app.chat_widget.capture_thread_input_state().unwrap();
// Recovered follow-ups stay paused until the user explicitly submits them.
input_state.recovered_queue = true;
app.chat_widget.restore_thread_input_state(
Some(input_state),
crate::chatwidget::ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
},
);
if target.thread_id != targets[0].thread_id {
let mut input_state = app.chat_widget.capture_thread_input_state().unwrap();
// Recovered follow-ups stay paused until the user explicitly submits them.
input_state.recovered_queue = true;
app.chat_widget.restore_thread_input_state(
Some(input_state),
crate::chatwidget::ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
},
);
}
expected_states.push(app.chat_widget.capture_thread_input_state());
} else {
let index = targets
@@ -2676,9 +2849,96 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> {
app.chat_widget.capture_thread_input_state(),
expected_states[index]
);
if target.thread_id == targets[0].thread_id {
app.chat_widget.apply_external_edit(steer_text.to_string());
app.chat_widget
.handle_key_event(KeyEvent::from(KeyCode::Enter));
let steer_event = std::iter::from_fn(|| events.try_recv().ok())
.find(|event| matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. })))
.expect("steer submission");
// Leave the steer uncommitted when exercising Esc's retry path.
if !interrupted {
Box::pin(app.handle_event(&mut tui, &mut server, steer_event)).await?;
}
app.chat_widget
.apply_external_edit(format!("Draft for {}", target.thread_id));
}
}
}
assert_eq!(app.chat_widget.thread_id(), Some(targets[1].thread_id));
if let Some(release) = release {
let outgoing = targets[0].thread_id;
assert_eq!(
app.agents_overview.input_states[&outgoing]
.pending_steers
.len(),
1
);
tokio::time::timeout(
Duration::from_secs(/*secs*/ 5),
responses_server.wait_for_request_count(/*count*/ 2),
)
.await?;
// A is unsubscribed now. Leave the completion out of the TUI event queue so
// resuming must discover it from the server's thread history.
release.send(()).unwrap();
wait_for_turn_status(&mut server, outgoing, TurnStatus::Completed).await?;
}
assert_eq!(
responses_server.requests().await.len(),
requests_before_return
);
while let Ok(event) = events.try_recv() {
assert!(!matches!(
event,
AppEvent::CodexOp(AppCommand::UserTurn { .. })
));
}
Box::pin(app.handle_event(
&mut tui,
&mut server,
AppEvent::ResumeSessionByIdOrName(targets[0].thread_id.to_string()),
))
.await?;
assert_eq!(app.active_thread_id, Some(targets[0].thread_id));
let mut submitted = Vec::new();
while let Ok(event) = events.try_recv() {
if let AppEvent::CodexOp(AppCommand::UserTurn { ref items, .. }) = event {
submitted.push(items.clone());
Box::pin(app.handle_event(&mut tui, &mut server, event)).await?;
}
}
let follow_up = format!("Follow-up for {}", targets[0].thread_id);
let expected_text = if interrupted {
steer_text.to_string()
} else {
follow_up.clone()
};
assert_eq!(
submitted,
vec![vec![codex_app_server_protocol::UserInput::Text {
text: expected_text,
text_elements: Vec::new(),
}]]
);
tokio::time::timeout(
Duration::from_secs(/*secs*/ 5),
responses_server.wait_for_request_count(requests_before_return + 1),
)
.await?;
assert_eq!(
responses_server.requests().await.len(),
requests_before_return + 1
);
if interrupted {
assert_eq!(app.chat_widget.queued_user_message_texts(), [follow_up]);
}
assert_eq!(
app.chat_widget.composer_text_with_pending(),
format!("Draft for {}", targets[0].thread_id)
);
server.shutdown().await?;
responses_server.shutdown().await;
Ok(())
}
codex-rs/tui/src/app/backend_selection.rs
+9 -26
View File
@@ -491,6 +491,8 @@ impl App {
Ok(prepared) => prepared,
Err(control) => return Ok(control),
};
let handoff = self.chat_widget.pause_input_for_resume_handoff();
let _handoff = self.prepare_resume_handoff(tui, handoff).await?;
self.switch_backend_context(
backends,
destination_slot,
@@ -500,35 +502,16 @@ impl App {
bootstrap,
)
.await;
let thread_id = target.thread_id;
let confirmed = super::thread_events::confirmed_user_message_ids(
&prepared.thread.turns,
std::iter::empty(),
);
let restored_blank_session = prepared.restored_blank_session;
let has_resumed_mode = prepared.thread.session.collaboration_mode.is_some();
backends.switch_displayed(destination_slot, &mut self.agents_overview.input_states);
let destination = backends.session_at_mut(destination_slot);
let control = self
.activate_prepared_session(tui, destination, prepared)
.activate_prepared_session_after_clear(
tui,
destination,
prepared,
/*switched_backend*/ true,
)
.await?;
backends.switch_displayed(destination_slot, &mut self.agents_overview.input_states);
if !restored_blank_session {
self.agents_overview.blank_sessions.remove(&thread_id);
}
if let Some(mut input) = self.agents_overview.input_states.remove(&thread_id) {
input.provider_error = None;
input
.pending_steers
.retain(|steer| !confirmed.contains(&steer.client_id));
if !restored_blank_session {
self.chat_widget
.update_input_collaboration_mode(&mut input, has_resumed_mode);
}
let settings = restored_blank_session
.then(|| input.pending_thread_settings.take())
.flatten();
self.restore_session_draft(thread_id, input, settings).await;
}
tui.frame_requester().schedule_frame();
Ok(control)
})
codex-rs/tui/src/app/session_lifecycle.rs
+144 -26
View File
@@ -33,6 +33,8 @@ pub(super) enum ThreadAttachPresentation {
/// A fresh thread whose startup composer is already visible.
FreshWithDraft,
SessionLineage,
/// A resume whose outgoing events were drained and terminal cleared before capture.
SessionLineageAfterClear,
}
pub(super) struct PreparedFreshSession {
@@ -1474,6 +1476,16 @@ impl App {
.await
}
async fn drain_outgoing_thread_events(&mut self) {
if let Some(mut receiver) = self.active_thread_rx.take() {
while let Ok(event) = receiver.try_recv() {
self.handle_thread_event_now_recovering_file_changes(event)
.await;
}
self.active_thread_rx = Some(receiver);
}
}
async fn replace_chat_widget_with_account(
&mut self,
tui: &mut tui::Tui,
@@ -1485,18 +1497,20 @@ impl App {
// Initial messages are for freshly attached primary threads only. Thread switches and
// resume/fork flows pass `None` so they cannot replay old history and then auto-submit a new
// user turn by accident.
if let Some(mut receiver) = self.active_thread_rx.take() {
while let Ok(event) = receiver.try_recv() {
self.handle_thread_event_now_recovering_file_changes(event)
.await;
}
self.active_thread_rx = Some(receiver);
if !matches!(
presentation,
ThreadAttachPresentation::SessionLineageAfterClear
) {
self.drain_outgoing_thread_events().await;
}
self.store_active_thread_receiver().await;
self.reset_transcript_state_after_clear();
tui.clear_pending_history_lines();
let presentation = match presentation {
ThreadAttachPresentation::FreshAfterClear => ThreadAttachPresentation::Fresh,
ThreadAttachPresentation::SessionLineageAfterClear => {
ThreadAttachPresentation::SessionLineage
}
ThreadAttachPresentation::FreshWithDraft => {
tui.defer_thread_switch_clear();
presentation
@@ -1742,6 +1756,43 @@ impl App {
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
prepared: super::resume_preparation::PreparedResume,
) -> Result<AppRunControl> {
if prepared.backend_id != app_server.backend_id() {
self.add_session_picker_error(
"This session belongs to a different backend. Open the session picker again."
.to_string(),
);
return Ok(AppRunControl::Continue);
}
let handoff = self.chat_widget.pause_input_for_resume_handoff();
let _handoff = self.prepare_resume_handoff(tui, handoff).await?;
self.activate_prepared_session_after_clear(
tui, app_server, prepared, /*switched_backend*/ false,
)
.await
}
pub(super) async fn prepare_resume_handoff(
&mut self,
tui: &mut tui::Tui,
handoff: std::sync::Arc<()>,
) -> Result<std::sync::Arc<()>> {
self.drain_outgoing_thread_events().await;
// Keep the outgoing receiver and session in place if the terminal clear fails.
if let Err(error) = tui.clear_for_thread_switch() {
drop(handoff);
self.chat_widget.maybe_send_next_queued_input();
return Err(error.into());
}
Ok(handoff)
}
pub(super) async fn activate_prepared_session_after_clear(
&mut self,
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
prepared: super::resume_preparation::PreparedResume,
switched_backend: bool,
) -> Result<AppRunControl> {
let super::resume_preparation::PreparedResume {
backend_id,
@@ -1751,24 +1802,19 @@ impl App {
read_only,
history_notice,
provider_error,
..
restored_blank_session,
} = prepared;
if backend_id != app_server.backend_id() {
self.add_session_picker_error(
"This session belongs to a different backend. Open the session picker again."
.to_string(),
);
return Ok(AppRunControl::Continue);
}
self.backend_registry
.observe(backend_id, resumed.session.thread_id);
debug_assert_eq!(backend_id, app_server.backend_id());
let resumed_thread_id = resumed.session.thread_id;
let was_reconnecting = self.chat_widget.thread_stream_disconnected;
self.backend_registry.observe(backend_id, resumed_thread_id);
let has_resumed_mode = resumed.session.collaboration_mode.is_some();
let was_reconnecting = !switched_backend && self.chat_widget.thread_stream_disconnected;
let owned_channel = self
.thread_event_channels
.get(&resumed_thread_id)
.filter(|channel| channel.backend_id.is_none_or(|owner| owner == backend_id));
let same_thread = self.chat_widget.thread_id() == Some(resumed_thread_id);
let same_thread =
!switched_backend && self.chat_widget.thread_id() == Some(resumed_thread_id);
let refreshing_history = same_thread
&& !was_reconnecting
&& owned_channel.is_some_and(|channel| channel.history_may_be_incomplete);
@@ -1780,9 +1826,37 @@ impl App {
|| self.chat_widget.has_provider_error()))
.then(|| self.chat_widget.capture_thread_input_state())
.flatten();
let confirmed_message_ids = refreshing_history.then(|| {
super::thread_events::confirmed_user_message_ids(&resumed.turns, std::iter::empty())
});
let mut confirmed_message_ids =
super::thread_events::confirmed_user_message_ids(&resumed.turns, std::iter::empty());
if let Some(channel) = owned_channel {
let store = channel.store.lock().await;
confirmed_message_ids.extend(super::thread_events::confirmed_user_message_ids(
&store.turns,
store.buffer.iter(),
));
}
// A backend switch has already parked the outgoing inputs.
if !switched_backend {
let displayed_thread_id = self.current_displayed_thread_id();
for (id, channel) in &self.thread_event_channels {
if Some(*id) != displayed_thread_id
&& channel.backend_id.is_none_or(|owner| owner == backend_id)
&& let Some(input) = channel.store.lock().await.input_state.clone()
{
self.agents_overview.input_states.insert(*id, input);
}
}
if displayed_thread_id != Some(resumed_thread_id)
&& let Some(id) = displayed_thread_id
&& self
.backend_registry
.owner(id)
.is_none_or(|owner| owner == backend_id)
&& let Some(input) = self.chat_widget.capture_thread_input_state()
{
self.agents_overview.input_states.insert(id, input);
}
}
self.detach_current_thread_for_navigation(app_server, Some(resumed_thread_id))
.await;
self.local_settings = local_settings;
@@ -1798,7 +1872,7 @@ impl App {
.replace_chat_widget_with_app_server_thread(
tui,
resumed,
ThreadAttachPresentation::SessionLineage,
ThreadAttachPresentation::SessionLineageAfterClear,
/*initial_user_message*/ None,
)
.await
@@ -1807,21 +1881,58 @@ impl App {
self.ensure_thread_channel(resumed_thread_id)
.history_may_be_incomplete = false;
self.remember_attached_host_project(app_server);
if let Some(mut input) = retained_input {
let resumed_turn_running = !read_only
&& self
.active_turn_id_for_thread(resumed_thread_id)
.await
.is_some();
let preserve_in_flight_turn = resumed_turn_running;
if switched_backend && !restored_blank_session {
self.agents_overview
.blank_sessions
.remove(&resumed_thread_id);
}
let keep_blank_settings = self
.agents_overview
.blank_sessions
.contains_key(&resumed_thread_id);
if let Some(mut input) =
retained_input.or(self.agents_overview.input_states.remove(&resumed_thread_id))
{
// A successful resume is the authority for clearing the old rejection.
input.provider_error = None;
if let Some(confirmed_message_ids) = confirmed_message_ids {
input.reconnect_pending = true;
if !restored_blank_session {
self.chat_widget
.update_input_collaboration_mode(&mut input, has_resumed_mode);
}
// A saved draft includes model settings, so apply newer server settings after
// restoring input for retained blank sessions.
let pending_settings = keep_blank_settings
.then(|| input.pending_thread_settings.take())
.flatten();
let reconcile_input = refreshing_history
|| !input.pending_steers.is_empty()
|| input.has_unconfirmed_messages();
if reconcile_input {
input.reconcile_confirmed_messages(&confirmed_message_ids);
input.reconnect_pending |= refreshing_history || !resumed_turn_running;
}
if refreshing_history
|| (reconcile_input && !read_only && !resumed_turn_running)
{
self.chat_widget
.restore_reconnected_input(Some(input), &confirmed_message_ids);
} else {
self.chat_widget.restore_thread_input_state(
Some(input),
ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
preserve_in_flight_turn,
},
);
}
if let Some(settings) = pending_settings {
self.chat_widget.on_thread_settings_updated(settings);
}
if was_reconnecting && !read_only && provider_error.is_none() {
self.chat_widget.maybe_send_next_queued_input();
}
@@ -1864,6 +1975,13 @@ impl App {
resumed_thread_id,
)
.await;
if self
.active_turn_id_for_thread(resumed_thread_id)
.await
.is_none()
{
self.chat_widget.maybe_send_next_queued_input();
}
}
}
Err(err) => return Err(err),
codex-rs/tui/src/app/session_picker.rs
-23
View File
@@ -1,7 +1,6 @@
//! In-app session picker transitions, preserving per-thread input and the view on cancellation.
use super::*;
use crate::chatwidget::ThreadInputStateRestoreMode;
impl App {
pub(super) async fn capture_session_inputs(&mut self, owner: crate::backend_id::BackendId) {
@@ -93,34 +92,12 @@ impl App {
return Ok(AppRunControl::Continue);
}
let thread_id = target_session.thread_id;
let switching_threads = self.current_displayed_thread_id() != Some(thread_id);
if switching_threads {
self.capture_session_inputs(app_server.backend_id()).await;
}
if let AppRunControl::Exit(reason) = self
.resume_target_session(tui, app_server, target_session)
.await?
{
return Ok(AppRunControl::Exit(reason));
}
if switching_threads
&& self.current_displayed_thread_id() == Some(thread_id)
&& let Some(mut input_state) =
self.agents_overview.input_states.remove(&thread_id)
{
// The resumed widget already reflects the latest server result.
input_state.provider_error = None;
self.reconcile_saved_session_input(thread_id, &mut input_state)
.await;
let preserve_in_flight_turn =
self.active_turn_id_for_thread(thread_id).await.is_some();
self.chat_widget.restore_thread_input_state(
Some(input_state),
ThreadInputStateRestoreMode {
preserve_in_flight_turn,
},
);
}
if self.active_thread_id == Some(thread_id)
&& self
.chat_widget
codex-rs/tui/src/app/tests/active_reconnect_tests.rs
+1 -1
View File
@@ -799,7 +799,7 @@ pub(super) async fn drain_history(
| AppEvent::BeginThreadSwitchHistoryReplayBuffer
| AppEvent::EndInitialHistoryReplayBuffer
) {
app.handle_event(tui, session, event).await?;
crate::boxed_future!(app.handle_event(tui, session, event)).await?;
}
}
Ok(lines_to_single_string(
codex-rs/tui/src/app/tests/backend_ownership_tests.rs
+206
View File
@@ -4,6 +4,7 @@ use super::start_recording_app_server;
use super::*;
use crate::app_server_session::ResumeModelSettings;
use crate::backend_id::BackendId;
use crate::chatwidget::tests::helpers::assert_pending_steers_unadmitted;
use crate::session_start::SessionStartAction;
use crate::session_start::SessionStartConfig;
use crate::session_start::complete_session_start;
@@ -351,3 +352,208 @@ async fn prepared_resume_cannot_activate_through_another_backend() -> Result<()>
proxy.abort();
Ok(())
}
#[tokio::test]
async fn read_only_resume_reconciles_receipts_before_writable_retry() -> Result<()> {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let mut tui = crate::tui::test_support::make_test_tui()?;
let (mut server, _requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
let current_thread = ThreadId::new();
let mut current_session = test_thread_session(current_thread, app.config.cwd.to_path_buf());
current_session.model = app.config.model.clone().expect("configured model");
current_session.model_provider_id = app.config.model_provider_id.clone();
app.enqueue_primary_thread_session(current_session, Vec::new())
.await?;
app.chat_widget.handle_server_notification(
ServerNotification::TurnStarted(codex_app_server_protocol::TurnStartedNotification {
thread_id: current_thread.to_string(),
turn: test_turn("current", TurnStatus::InProgress, Vec::new()),
}),
/*replay_kind*/ None,
);
app.chat_widget
.submit_user_message_as_plain_user_turn(crate::chatwidget::UserMessage::from(
"accepted steer",
))
.expect("submit steer");
app.chat_widget
.submit_user_message_as_plain_user_turn(crate::chatwidget::UserMessage::from(
"delayed steer",
))
.expect("submit delayed steer");
app.chat_widget
.apply_external_edit("retained draft".to_string());
let input = app
.chat_widget
.capture_thread_input_state()
.expect("pending input");
assert_eq!(input.pending_steers.len(), 2);
assert!(!input.reconnect_pending);
assert_pending_steers_unadmitted(&input);
let command_center_input = input.clone();
let accepted_client_id = input.pending_steers[0].client_id.clone();
let delayed_client_id = input.pending_steers[1].client_id.clone();
let resumed_thread = ThreadId::new();
app.agents_overview
.input_states
.insert(resumed_thread, input);
let receipt = |id: &str, client_id: String, text: &str| ThreadItemPayload::UserMessage {
id: id.into(),
client_id: Some(client_id),
content: vec![UserInput::Text {
text: text.into(),
text_elements: Vec::new(),
}],
};
let accepted_steer = receipt("accepted-steer", accepted_client_id, "accepted steer");
let delayed_steer = receipt("delayed-steer", delayed_client_id, "delayed steer");
let backend_id = server.backend_id();
let resumed_mode = CollaborationMode {
mode: ModeKind::Plan,
settings: Settings {
model: "resumed-model".into(),
reasoning_effort: Some(ReasoningEffortConfig::High),
developer_instructions: None,
},
};
let prepared = |app: &App, read_only, status, items| {
let mut session = test_thread_session(resumed_thread, app.config.cwd.to_path_buf());
session.model = resumed_mode.model().to_string();
session.reasoning_effort = resumed_mode.reasoning_effort();
session.collaboration_mode = Some(Box::new(resumed_mode.clone()));
session.model_provider_id = app.config.model_provider_id.clone();
super::super::resume_preparation::PreparedResume {
backend_id,
config: app.config.clone(),
local_settings: app.local_settings.clone(),
thread: AppServerStartedThread {
session,
turns: vec![test_turn("resumed", status, items)],
blocks_direct_input: false,
task_tools_available: true,
},
restored_blank_session: false,
read_only,
history_notice: None,
provider_error: None,
}
};
// Exercise the Command Center restore boundary with the raw saved input,
// rather than the uncertainty already marked by direct resume.
let idle_resume = prepared(
&app,
/*read_only*/ false,
TurnStatus::Completed,
vec![accepted_steer.clone()],
);
app.activate_prepared_session(&mut tui, &mut server, idle_resume)
.await?;
assert_eq!(app.active_turn_id_for_thread(resumed_thread).await, None);
assert!(!app.chat_widget.is_external_writer_view());
app.restore_session_draft(
resumed_thread,
command_center_input.clone(),
/*pending_settings*/ None,
)
.await;
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"retained draft"
);
assert_eq!(
app.chat_widget.queued_user_message_texts(),
["delayed steer"]
);
assert!(
app.chat_widget
.capture_thread_input_state()
.unwrap()
.has_unconfirmed_messages()
);
for item in [
receipt("wrong-id", "unrelated-client-id".into(), "delayed steer"),
delayed_steer.clone(),
] {
assert_eq!(
app.chat_widget.queued_user_message_texts(),
["delayed steer"]
);
app.chat_widget.handle_server_notification(
ServerNotification::ItemCompleted(
codex_app_server_protocol::ItemCompletedNotification::new(
resumed_thread.to_string(),
"resumed".into(),
item,
/*started_at_ms*/ None,
/*completed_at_ms*/ 1,
),
),
/*replay_kind*/ None,
);
}
assert!(app.chat_widget.queued_user_message_texts().is_empty());
app.agents_overview
.input_states
.insert(resumed_thread, command_center_input);
let read_only_resume = prepared(
&app,
/*read_only*/ true,
TurnStatus::Completed,
vec![accepted_steer.clone()],
);
app.activate_prepared_session(&mut tui, &mut server, read_only_resume)
.await?;
assert!(app.chat_widget.is_external_writer_view());
assert_eq!(app.chat_widget.effective_collaboration_mode(), resumed_mode);
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"retained draft"
);
assert_eq!(
app.chat_widget.queued_user_message_texts(),
vec!["delayed steer"]
);
let input = app
.chat_widget
.capture_thread_input_state()
.expect("unconfirmed steer");
assert!(input.has_unconfirmed_messages());
let writable_resume = prepared(
&app,
/*read_only*/ false,
TurnStatus::InProgress,
vec![accepted_steer, delayed_steer],
);
app.activate_prepared_session(&mut tui, &mut server, writable_resume)
.await?;
assert!(!app.chat_widget.is_external_writer_view());
assert!(app.chat_widget.is_agent_turn_running());
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"retained draft"
);
assert!(app.chat_widget.queued_user_message_texts().is_empty());
let input = app
.chat_widget
.capture_thread_input_state()
.expect("retained draft");
assert!(input.pending_steers.is_empty());
while let Ok(event) = events.try_recv() {
assert!(!matches!(
event,
AppEvent::CodexOp(AppCommand::UserTurn { .. })
));
}
server.shutdown().await?;
proxy.abort();
Ok(())
}
codex-rs/tui/src/app/tests/history_hydration_tests.rs
+2 -2
View File
@@ -222,7 +222,7 @@ async fn history_hydration_metadata_tracks_missing_turns_and_stale_completions_k
backwards_cursor: None,
}),
] {
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::OlderThreadHistoryLoaded {
@@ -230,7 +230,7 @@ async fn history_hydration_metadata_tracks_missing_turns_and_stale_completions_k
cursor: old_cursor.clone(),
result,
},
)
))
.await?;
assert!(server.is_older_history_page_pending(target.thread_id, &cursor));
assert_eq!(
codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs
+11 -7
View File
@@ -331,13 +331,13 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
)
.await?;
let old_event_tx = app.app_event_tx.app_event_tx.clone();
app.finish_reconnect(
crate::boxed_future!(app.finish_reconnect(
&mut tui,
&mut session,
&mut events,
connected,
CODEX_CLI_VERSION,
)
))
.await?;
assert_eq!(app.app_server_backend_id, Some(session.backend_id()));
assert!(old_event_tx.is_closed());
@@ -373,7 +373,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
while app.agents_overview.active_refresh.is_some() {
let event = events.recv().await.expect("inventory refresh event");
if matches!(event, AppEvent::AgentsOverviewThreadsLoaded { .. }) {
Box::pin(app.handle_event(&mut tui, &mut session, event)).await?;
crate::boxed_future!(app.handle_event(&mut tui, &mut session, event)).await?;
} else {
deferred.push(event);
}
@@ -507,8 +507,12 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
app.chat_widget.composer_text_with_pending(),
"latest draft!"
);
Box::pin(app.handle_event(&mut tui, &mut session, AppEvent::OpenAgentsOverview))
.await?;
crate::boxed_future!(app.handle_event(
&mut tui,
&mut session,
AppEvent::OpenAgentsOverview
))
.await?;
assert!(app.chat_widget.has_active_view());
app.handle_tui_event(
&mut tui,
@@ -554,7 +558,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
);
}
available.store(true, std::sync::atomic::Ordering::SeqCst);
Box::pin(app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut session,
AppEvent::SelectAgentsOverviewThread { thread_id: id },
@@ -604,7 +608,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv
.await?;
while let Ok(event) = events.try_recv() {
if matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. })) {
Box::pin(app.handle_event(&mut tui, &mut session, event)).await?;
crate::boxed_future!(app.handle_event(&mut tui, &mut session, event)).await?;
}
}
let channel = app.thread_event_channels.get(&id).unwrap();
codex-rs/tui/src/app/tests/new_session_tests.rs
+13
View File
@@ -1199,6 +1199,9 @@ async fn check_retained_draft_handoff(
let mode = server_mode.as_mut().expect("test has a mode to update");
mode.mode = ModeKind::Plan;
mode.settings.developer_instructions = Some("Instructions from the parked task".into());
if parked_policy.is_some() {
mode.settings.reasoning_effort = Some(ReasoningEffortConfig::High);
}
Some(mode.clone())
} else {
None
@@ -1220,6 +1223,7 @@ async fn check_retained_draft_handoff(
let settings = ThreadSettingsUpdateParams {
thread_id: thread_id.to_string(),
collaboration_mode,
effort: parked_policy.map(|_| ReasoningEffortConfig::High),
approval_policy: parked_policy,
..Default::default()
};
@@ -1236,6 +1240,15 @@ async fn check_retained_draft_handoff(
let (destination, parked_inputs) = backends.session_and_parked_inputs(destination_slot);
app.handle_polled_backend_event(destination, outgoing, parked_inputs, event)
.await;
} else if saved_history && retained_blank {
// A formerly blank thread must ignore settings cached before it acquired history.
let mut stale_update = update;
stale_update.thread_settings.model = "old-model".into();
backends
.parked_inputs_mut(destination_slot)
.get_mut(&thread_id)
.unwrap()
.pending_thread_settings = Some(stale_update);
}
}
let settings = app.backend_settings();
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+13 -13
View File
@@ -1133,7 +1133,7 @@ async fn complete_managed_worktree_creation(
.await?
.ok_or_else(|| color_eyre::eyre::eyre!("worktree creation event channel closed"))?;
if matches!(event, AppEvent::ManagedWorktreeCreated(_)) {
app.handle_event(tui, server, event).await?;
crate::boxed_future!(app.handle_event(tui, server, event)).await?;
if let Some(created) = app.pending_managed_worktree_created.take() {
app.finish_managed_worktree(*created).await;
}
@@ -4796,14 +4796,14 @@ model_reasoning_effort = "low"
);
let mut browser_entries = Vec::new();
requests.lock().expect("request recorder lock").clear();
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::StartManagedWorktree {
mode: ManagedWorktreeMode::Fork,
name: None,
},
)
))
.await?;
complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?;
assert_eq!(app.chat_widget.thread_id(), Some(original));
@@ -4869,14 +4869,14 @@ terminal_visualization_instructions = true
)?;
app.config = app.rebuild_config_for_cwd(source.clone()).await?;
requests.lock().expect("request recorder lock").clear();
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::StartManagedWorktree {
mode: ManagedWorktreeMode::Fork,
name: None,
},
)
))
.await?;
complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?;
assert_eq!(app.chat_widget.thread_id(), Some(original));
@@ -4893,14 +4893,14 @@ terminal_visualization_instructions = true
owner: crate::worktree_browser::Owner::None,
});
while events.try_recv().is_ok() {}
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::StartManagedWorktree {
mode: ManagedWorktreeMode::New,
name: None,
},
)
))
.await?;
assert_eq!(app.chat_widget.thread_id(), Some(original));
drain_managed_worktree_start(&mut app, &mut server).await;
@@ -4928,11 +4928,11 @@ terminal_visualization_instructions = true
.cwd
.clone();
app.primary_thread_id = Some(ThreadId::new());
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::ManagedWorktreeCreated(created),
)
))
.await?;
let created = app
.pending_managed_worktree_created
@@ -4968,14 +4968,14 @@ terminal_visualization_instructions = true
requests.lock().expect("request recorder lock").clear();
let previous = app.chat_widget.thread_id();
let previous_rollout = app.chat_widget.rollout_path().expect("previous rollout");
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::StartManagedWorktree {
mode,
name: Some(format!("{mode:?} worktree")),
},
)
))
.await?;
complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?;
let replacement = app.chat_widget.thread_id().expect("replacement thread");
@@ -5146,11 +5146,11 @@ terminal_visualization_instructions = true
.set_turns(vec![completed]);
for mode in [ManagedWorktreeMode::Fork, ManagedWorktreeMode::New] {
requests.lock().expect("request recorder lock").clear();
app.handle_event(
crate::boxed_future!(app.handle_event(
&mut tui,
&mut server,
AppEvent::StartManagedWorktree { mode, name: None },
)
))
.await?;
drain_managed_worktree_start(&mut app, &mut server).await;
let is_new = mode == ManagedWorktreeMode::New;
codex-rs/tui/src/app/tests/unavailable_commands_tests.rs
+53
View File
@@ -4,6 +4,59 @@ use super::active_reconnect::drain_history;
use super::*;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn cloud_shell_command_reports_error_and_allows_next_prompt() -> Result<()> {
let (mut app, mut events, mut ops) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
app.active_thread_id = Some(thread_id);
app.app_server_target = AppServerTarget::Cloud;
app.chat_widget
.handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf()));
let (client, mut requests, _server_events) =
codex_app_server_client::CloudClient::channel(/*capacity*/ 1);
let mut session = AppServerSession::new(
crate::backend_id::BackendId::new(),
codex_app_server_client::AppServerClient::Cloud(client),
crate::app_server_session::ThreadParamsMode::Cloud,
);
let mut tui = crate::tui::test_support::make_test_tui()?;
while events.try_recv().is_ok() {}
while ops.try_recv().is_ok() {}
app.chat_widget
.restore_user_message_to_composer("First prompt".into());
app.chat_widget.handle_key_event(KeyCode::Enter.into());
assert!(matches!(ops.try_recv()?, Op::UserTurn { .. }));
app.chat_widget.handle_server_notification(
turn_started_notification(thread_id, "turn-1"),
/*replay_kind*/ None,
);
app.chat_widget.handle_server_notification(
turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed),
/*replay_kind*/ None,
);
while events.try_recv().is_ok() {}
app.chat_widget
.restore_user_message_to_composer("! pwd".into());
app.chat_widget.handle_key_event(KeyCode::Enter.into());
let shell = ops.try_recv()?;
assert!(matches!(&shell, Op::RunUserShellCommand { command } if command == "pwd"));
let result = tokio::select! {
result = Box::pin(app.handle_event(&mut tui, &mut session, AppEvent::CodexOp(shell))) => result?,
_ = requests.recv() => panic!("Cloud shell command unexpectedly sent a request"),
};
assert!(matches!(result, AppRunControl::Continue));
assert_snapshot!(next_history_message(&mut events), @"■ Shell commands starting with ! are unavailable in Cloud conversations.");
assert_eq!(app.active_thread_id, Some(thread_id));
app.chat_widget
.restore_user_message_to_composer("Continue the conversation".into());
app.chat_widget.handle_key_event(KeyCode::Enter.into());
assert!(matches!(ops.try_recv()?, Op::UserTurn { .. }));
Ok(())
}
#[tokio::test]
async fn unavailable_thread_routes_local_and_recovery_commands() -> Result<()> {
let (mut app, mut events, mut ops) = make_test_app_with_channels().await;
codex-rs/tui/src/app/thread_routing.rs
+11 -1
View File
@@ -970,6 +970,15 @@ impl App {
Ok(true)
}
AppCommand::RunUserShellCommand { command } => {
if app_server.thread_params_mode()
== crate::app_server_session::ThreadParamsMode::Cloud
{
self.chat_widget.reject_user_shell_command(
"Shell commands starting with ! are unavailable in Cloud conversations."
.into(),
);
return Ok(true);
}
app_server
.thread_shell_command(thread_id, command.to_string())
.await?;
@@ -1613,7 +1622,8 @@ impl App {
ThreadAttachPresentation::Fresh
| ThreadAttachPresentation::FreshAfterClear
| ThreadAttachPresentation::FreshWithDraft
| ThreadAttachPresentation::SessionLineage => {
| ThreadAttachPresentation::SessionLineage
| ThreadAttachPresentation::SessionLineageAfterClear => {
self.chat_widget.handle_thread_session(session);
}
}
codex-rs/tui/src/chatwidget/input_queue.rs
+5 -1
View File
@@ -50,6 +50,8 @@ pub(super) struct InputQueueState {
/// fresh user turn instead of restoring them into the composer.
pub(super) submit_pending_steers_after_interrupt: bool,
pub(super) suppress_queue_autosend: bool,
/// Independent of settings suppression, which a drained event can set.
pub(super) resume_handoff: std::sync::Weak<()>,
pub(super) transcript_copy: std::sync::Weak<crate::copy_input_guard::CopyInputGuard>,
/// Hold submissions while a usage failure or backend-directed model fallback is resolved.
pub(super) rate_limit_recovery_pending: bool,
@@ -67,7 +69,9 @@ impl InputQueueState {
}
pub(super) fn submissions_paused(&self) -> bool {
self.suppress_queue_autosend || self.transcript_copy.strong_count() > 0
self.suppress_queue_autosend
|| self.transcript_copy.strong_count() > 0
|| self.resume_handoff.strong_count() > 0
}
pub(super) fn has_unconfirmed_messages(&self) -> bool {
codex-rs/tui/src/chatwidget/input_restore.rs
+42 -14
View File
@@ -462,6 +462,7 @@ impl ChatWidget {
// interrupted turn reaches the UI, so any unacknowledged steers still
// tracked here must be restored locally instead of waiting for a later commit.
if send_pending_steers_immediately {
let resume_handoff = self.input_queue.resume_handoff.strong_count() > 0;
let pending_steers = self
.input_queue
.pending_steers
@@ -483,7 +484,9 @@ impl ChatWidget {
.rejected_steer_history_records
.push_back(pending.history_record);
}
if let Some((message, history_record)) = self.pop_next_queued_user_message() {
if !resume_handoff
&& let Some((message, history_record)) = self.pop_next_queued_user_message()
{
let source = message.source;
self.submit_user_message_with_history_and_shell_escape_policy(
message.into_user_message(),
@@ -499,12 +502,22 @@ impl ChatWidget {
.map(|pending| (pending.user_message, pending.history_record))
.collect(),
);
self.submit_user_message_with_history_and_shell_escape_policy(
user_message,
history_record,
ShellEscapePolicy::Allow,
UserMessageSource::Prompt,
);
if resume_handoff {
// Capture the retry with the outgoing thread's input.
self.input_queue
.queued_user_messages
.push_front(user_message.into());
self.input_queue
.queued_user_message_history_records
.push_front(history_record);
} else {
self.submit_user_message_with_history_and_shell_escape_policy(
user_message,
history_record,
ShellEscapePolicy::Allow,
UserMessageSource::Prompt,
);
}
} else if let Some(combined) = self.drain_pending_messages_for_restore() {
self.restore_composer_state(combined);
}
@@ -634,7 +647,8 @@ impl ChatWidget {
}
pub(super) fn restore_composer_state(&mut self, composer: ThreadComposerState) {
self.restore_composer_state_with_shell_mode(composer, /*shell_mode*/ None);
let shell_mode = composer.shell_mode;
self.restore_composer_state_with_shell_mode(composer, shell_mode);
}
fn restore_composer_state_with_shell_mode(
@@ -644,6 +658,7 @@ impl ChatWidget {
) {
let ThreadComposerState {
text,
shell_mode: _,
local_images,
remote_image_urls,
text_elements,
@@ -679,6 +694,7 @@ impl ChatWidget {
} = user_message;
ThreadComposerState {
text,
shell_mode: None,
local_images,
remote_image_urls,
text_elements,
@@ -692,6 +708,8 @@ impl ChatWidget {
let draft = self.bottom_pane.composer_recovery_snapshot();
let composer = ThreadComposerState {
text: draft.text,
// A disabled shell composer also treats a leading bang literally.
shell_mode: Some(draft.shell_mode.unwrap_or(false)),
text_elements: draft.text_elements,
local_images: draft.local_images,
remote_image_urls: draft.remote_image_urls,
@@ -763,8 +781,14 @@ impl ChatWidget {
.as_ref()
.and_then(|state| state.provider_error.clone());
let preserve_in_flight_turn = restore_mode.preserve_in_flight_turn;
let restored_task_running =
preserve_in_flight_turn && input_state.as_ref().is_some_and(|state| state.task_running);
let restored_task_running = preserve_in_flight_turn
&& (self.bottom_pane.is_task_running()
|| input_state.as_ref().is_some_and(|state| state.task_running));
let restored_agent_turn_running = preserve_in_flight_turn
&& (self.turn_lifecycle.agent_turn_running
|| input_state
.as_ref()
.is_some_and(|state| state.agent_turn_running));
if let Some(input_state) = input_state {
self.bottom_pane.restore_questions(input_state.questions);
self.initial_user_message = input_state.initial_user_message;
@@ -782,10 +806,8 @@ impl ChatWidget {
self.restore_status_surface_overrides(input_state.status_surface_overrides);
self.safety_buffering_prompt = input_state.safety_buffering_prompt;
self.safety_buffering_source = input_state.safety_buffering_source;
self.turn_lifecycle.restore_running(
preserve_in_flight_turn && input_state.agent_turn_running,
Instant::now(),
);
self.turn_lifecycle
.restore_running(restored_agent_turn_running, Instant::now());
self.input_queue.user_turn_pending_start =
preserve_in_flight_turn && input_state.user_turn_pending_start;
self.input_queue.pending_user_message_client_id =
@@ -864,6 +886,12 @@ impl ChatWidget {
self.request_redraw();
}
pub(crate) fn pause_input_for_resume_handoff(&mut self) -> std::sync::Arc<()> {
let handoff = std::sync::Arc::new(());
self.input_queue.resume_handoff = std::sync::Arc::downgrade(&handoff);
handoff
}
pub(crate) fn set_queue_autosend_suppressed(&mut self, suppressed: bool) {
self.input_queue.suppress_queue_autosend = suppressed;
}
codex-rs/tui/src/chatwidget/input_submission.rs
+7
View File
@@ -51,6 +51,13 @@ impl ChatWidget {
}
}
pub(crate) fn reject_user_shell_command(&mut self, message: String) {
// No turn event will arrive to clear the pending shell submission.
self.input_queue.user_turn_pending_start = false;
self.add_error_message(message);
self.maybe_send_next_queued_input();
}
fn submit_shell_command_with_history(
&mut self,
command: &str,
codex-rs/tui/src/chatwidget/questions.rs
+1
View File
@@ -178,6 +178,7 @@ impl ChatWidget {
let cursor = main.cursor;
self.restore_composer_state(ThreadComposerState {
text: main.text,
shell_mode: None,
text_elements: main.text_elements,
local_images: main.local_images,
remote_image_urls: main.remote_image_urls,
codex-rs/tui/src/chatwidget/tests/composer_submission.rs
+34 -1
View File
@@ -1382,7 +1382,7 @@ async fn blocked_image_restore_with_remote_images_keeps_local_placeholder_mappin
}
#[tokio::test]
async fn startup_draft_handoff_preserves_literal_and_shell_bang() {
async fn draft_handoffs_preserve_literal_and_shell_bang() {
for text in [
"!echo literal",
"!echo literal\nsecond line",
@@ -1400,6 +1400,13 @@ async fn startup_draft_handoff_preserves_literal_and_shell_bang() {
chat.thread_id = Some(ThreadId::new());
chat.restore_startup_draft(draft.clone());
assert_eq!(chat.bottom_pane.composer_draft_snapshot(), draft);
let input = chat.capture_thread_input_state();
chat.restore_thread_input_state(
input,
ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
},
);
chat.handle_key_event(KeyEvent::from(KeyCode::Enter));
assert_hidden_shell_payload_is_literal(ops.try_recv(), text.trim().to_string());
}
@@ -1413,6 +1420,13 @@ async fn startup_draft_handoff_preserves_literal_and_shell_bang() {
let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.restore_startup_draft(draft);
let input = chat.capture_thread_input_state();
chat.restore_thread_input_state(
input,
ThreadInputStateRestoreMode {
preserve_in_flight_turn: false,
},
);
chat.handle_key_event(KeyEvent::from(KeyCode::Enter));
assert_matches!(ops.try_recv(), Ok(Op::RunUserShellCommand { command }) if command == "pwd");
}
@@ -2744,7 +2758,26 @@ async fn reconnect_resumes_unsent_input_and_reconciles_confirmed_submissions() {
let mut input = chat.capture_thread_input_state();
let saved_source = input.as_ref().unwrap().safety_buffering_source;
assert_eq!(saved_source, UserMessageSource::Prompt);
if !pending_start {
let mut running_input = input.clone().unwrap();
running_input.reconcile_confirmed_messages(std::slice::from_ref(&client_id));
let (mut restored, _rx, mut ops) =
make_chatwidget_manual(/*model_override*/ None).await;
restored.restore_thread_input_state(
Some(running_input),
ThreadInputStateRestoreMode {
preserve_in_flight_turn: true,
},
);
assert!(restored.turn_lifecycle.agent_turn_running);
assert!(restored.input_queue.pending_steers.is_empty());
assert!(!restored.input_queue.submit_pending_steers_after_interrupt);
restored.on_interrupted_turn(TurnAbortReason::Interrupted);
assert_no_submit_op(&mut ops);
}
// Missing history and unrelated receipts never authorize resubmitting uncertain input.
// Each pass restores state captured from the prior pass, so an exact receipt discovered
// on a later resume still releases the ordinary follow-up exactly once.
for ids in [
vec![],
vec!["different-submission".into()],
codex-rs/tui/src/chatwidget/tests/helpers.rs
+7
View File
@@ -6,6 +6,13 @@ use codex_app_server_protocol::PluginAvailability;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq;
#[track_caller]
pub(crate) fn assert_pending_steers_unadmitted(input: &ThreadInputState) {
for steer in &input.pending_steers {
assert_eq!(steer.admitted_turn_id, None);
}
}
pub(super) async fn test_config() -> (tempfile::TempDir, Config) {
// Start from the built-in defaults so tests do not inherit host/system config.
let codex_home = tempfile::Builder::new()
codex-rs/tui/src/chatwidget/user_messages.rs
+3 -1
View File
@@ -133,6 +133,7 @@ pub(super) enum QueueDrain {
#[derive(Debug, Clone, PartialEq, Default)]
pub(super) struct ThreadComposerState {
pub(super) text: String,
pub(super) shell_mode: Option<bool>,
pub(super) local_images: Vec<LocalImageAttachment>,
pub(super) remote_image_urls: Vec<String>,
pub(super) text_elements: Vec<TextElement>,
@@ -169,7 +170,7 @@ pub(crate) struct ThreadInputState {
pub(super) queued_user_messages: VecDeque<QueuedUserMessage>,
pub(super) queued_user_message_history_records: VecDeque<UserMessageHistoryRecord>,
pub(crate) recovered_queue: bool,
/// Reconcile delivery after transport recovery, not after a fork or ordinary thread switch.
/// Reconcile delivery against resumed history after recovery or an idle thread switch.
pub(crate) reconnect_pending: bool,
pub(super) user_turn_pending_start: bool,
pub(super) pending_user_message_client_id: Option<String>,
@@ -886,6 +887,7 @@ impl ThreadInputState {
pub(crate) fn reconcile_confirmed_messages(&mut self, confirmed_ids: &[String]) {
self.pending_steers
.retain(|steer| !confirmed_ids.contains(&steer.client_id));
self.submit_pending_steers_after_interrupt &= !self.pending_steers.is_empty();
remove_confirmed_queued_messages(
&mut self.queued_user_messages,
&mut self.queued_user_message_history_records,