From 54a4077c8b010bd36720804adb8301c1e6883650 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Wed, 2 Sep 2026 23:13:36 +0000 Subject: [PATCH] Preserve restored input after resolved misalignment errors (#42399) ## Why Replaying a thread after a later turn has continued can treat an earlier misalignment policy violation as still active, clearing the restored draft and queued follow-up input. ## What changed - Track the latest turn independently of the bounded replay buffer. - Omit misalignment errors from older turns during snapshot and stored-turn replay while preserving an error on the latest turn. - Keep interrupted tool calls from the older turn visible in history. ## Testing Added buffered replay coverage for restored drafts and queued messages, buffer eviction, current-turn violations, and interrupted tool rendering. GitOrigin-RevId: d6c223d9c5a1905fe1b48dc5f9514a7ccbf61d92 --- codex-rs/tui/src/app/replay_filter.rs | 31 +++ codex-rs/tui/src/app/tests/buffered_replay.rs | 178 ++++++++++++++++++ ...ay__misalignment_replayed_failed_tool.snap | 6 + ...d_replay__misalignment_replayed_input.snap | 11 ++ codex-rs/tui/src/app/thread_events.rs | 28 ++- codex-rs/tui/src/app/thread_routing.rs | 6 +- codex-rs/tui/src/chatwidget/replay.rs | 12 +- 7 files changed, 266 insertions(+), 6 deletions(-) create mode 100644 codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_failed_tool.snap create mode 100644 codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_input.snap diff --git a/codex-rs/tui/src/app/replay_filter.rs b/codex-rs/tui/src/app/replay_filter.rs index 707192cb06..0c2cbff6f7 100644 --- a/codex-rs/tui/src/app/replay_filter.rs +++ b/codex-rs/tui/src/app/replay_filter.rs @@ -1,5 +1,6 @@ //! Helpers for deciding which buffered events to replay when switching threads. +use codex_app_server_protocol::CodexErrorInfo; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ThreadItem; @@ -38,6 +39,36 @@ pub(super) fn event_is_notice(event: &ThreadBufferedEvent) -> bool { ) } +/// A later turn resolves an earlier precaution; replay must preserve restored input. +pub(super) fn omit_resolved_misalignment_errors( + snapshot: &mut ThreadEventSnapshot, + latest_turn: &str, +) { + snapshot.events.retain(|event| !matches!( + event, + ThreadBufferedEvent::Notification(notification) + if matches!(notification.as_ref(), ServerNotification::Error(n) + if n.turn_id != latest_turn + && n.error.codex_error_info == Some(CodexErrorInfo::MisalignmentPolicyViolation)) + )); + let completed_turns = snapshot.events.iter_mut().filter_map(|event| match event { + ThreadBufferedEvent::Notification(notification) => match notification.as_mut() { + ServerNotification::TurnCompleted(n) => Some(&mut n.turn), + _ => None, + }, + _ => None, + }); + for turn in snapshot.turns.iter_mut().chain(completed_turns) { + if turn.id != latest_turn + && turn.error.as_ref().is_some_and(|error| { + error.codex_error_info == Some(CodexErrorInfo::MisalignmentPolicyViolation) + }) + { + turn.error = None; + } + } +} + /// A completed item's full text replaces its earlier streaming deltas during replay. /// Keep deltas without a later completion so an in-progress or truncated stream still renders. /// Other events are barriers: streaming text may need to flush before a tool or prompt. diff --git a/codex-rs/tui/src/app/tests/buffered_replay.rs b/codex-rs/tui/src/app/tests/buffered_replay.rs index a45a20350e..31930bb752 100644 --- a/codex-rs/tui/src/app/tests/buffered_replay.rs +++ b/codex-rs/tui/src/app/tests/buffered_replay.rs @@ -1,6 +1,7 @@ //! Real buffered notifications must use finalized rendering without losing live stream tails. use super::*; +use crate::chatwidget::tests::helpers::normalize_snapshot_paths; use codex_app_server_protocol::ItemCompletedNotification; use pretty_assertions::assert_eq; @@ -120,3 +121,180 @@ fn buffered_replay_preserves_text_before_interactive_requests() { replay_filter::omit_completed_agent_deltas(&mut events); assert_eq!(format!("{events:?}"), before); } + +#[tokio::test] +async fn misalignment_buffered_replay_preserves_input_after_continuation() { + let (mut app, mut history, _) = make_test_app_with_channels().await; + let thread_id = ThreadId::new(); + app.chat_widget + .set_queue_autosend_suppressed(/*suppressed*/ true); + app.chat_widget.insert_str("keep queued"); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Tab)); + app.chat_widget + .restore_user_message_to_composer("keep draft".into()); + let saved_input = app.chat_widget.capture_thread_input_state(); + let error = AppServerTurnError { + misalignment: None, + message: "Chat stopped".into(), + codex_error_info: Some(AppServerCodexErrorInfo::MisalignmentPolicyViolation), + additional_details: None, + }; + let failed_turn = Turn { + error: Some(error.clone()), + items: vec![ThreadItem::McpToolCall { + id: "tool-call".into(), + server: "sample".into(), + tool: "inspect".into(), + status: codex_app_server_protocol::McpToolCallStatus::InProgress, + arguments: serde_json::json!({}), + app_context: None, + mcp_app_resource_uri: None, + plugin_id: None, + read_only_hint: None, + result: None, + error: None, + duration_ms: None, + }], + ..test_turn("failed-turn", TurnStatus::Failed, Vec::new()) + }; + let continued_turn = test_turn("continued-turn", TurnStatus::Completed, Vec::new()); + // Primary resume also replays stored turns without going through the snapshot filter. + app.chat_widget + .set_queue_autosend_suppressed(/*suppressed*/ true); + app.chat_widget.replay_thread_turns( + vec![failed_turn.clone(), continued_turn.clone()], + ReplayKind::ResumeInitialMessages, + ); + let rendered = std::iter::from_fn(|| history.try_recv().ok()) + .filter_map(|event| match event { + AppEvent::InsertHistoryCell(cell) => { + Some(lines_to_single_string(&cell.display_lines(/*width*/ 80))) + } + _ => None, + }) + .collect::(); + insta::assert_snapshot!("misalignment_replayed_failed_tool", rendered); + assert!(!app.chat_widget.has_misalignment_policy_violation()); + insta::assert_snapshot!( + "misalignment_replayed_input", + normalize_snapshot_paths(render_bottom_popup(&app.chat_widget, /*width*/ 80)) + ); + assert_eq!(app.chat_widget.composer_text_with_pending(), "keep draft"); + assert_eq!( + app.chat_widget.queued_user_message_texts(), + vec!["keep queued".to_string()] + ); + let events = [ + turn_started_notification(thread_id, "failed-turn"), + ServerNotification::Error(codex_app_server_protocol::ErrorNotification { + thread_id: thread_id.to_string(), + turn_id: "failed-turn".to_string(), + error, + will_retry: false, + }), + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: failed_turn.clone(), + }), + turn_started_notification(thread_id, "continued-turn"), + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: continued_turn, + }), + ]; + // Retain the whole stream, only the new turn's delta, or only a post-completion notice. + for (capacity, event_count) in [(16, 5), (1, 4), (1, 5)] { + let mut store = ThreadEventStore::new_with_session( + capacity, + test_thread_session(thread_id, app.config.cwd.to_path_buf()), + vec![failed_turn.clone()], + ); + store.input_state = saved_input.clone(); + for event in &events[..event_count] { + store.push_notification_ref(event); + } + if capacity == 1 { + store.push_notification(if event_count == 4 { + delta(&thread_id.to_string(), "continued-turn", "answer") + } else { + ServerNotification::Warning(WarningNotification { + message: "An unrelated notice".into(), + thread_id: Some(thread_id.to_string()), + }) + }); + assert_eq!(store.buffer.len(), 1); + } + app.replay_thread_snapshot(store.snapshot(), /*resume_restored_queue*/ false); + assert!(!app.chat_widget.has_misalignment_policy_violation()); + assert_eq!(app.chat_widget.composer_text_with_pending(), "keep draft"); + assert_eq!( + app.chat_widget.queued_user_message_texts(), + vec!["keep queued".to_string()] + ); + } +} + +#[tokio::test] +async fn misalignment_replay_blocks_when_turn_start_was_evicted() { + let thread_id = ThreadId::new(); + let error = AppServerTurnError { + misalignment: None, + message: "Chat stopped".into(), + codex_error_info: Some(AppServerCodexErrorInfo::MisalignmentPolicyViolation), + additional_details: None, + }; + let previous_turn = test_turn("old-turn", TurnStatus::Completed, Vec::new()); + let failed_turn = Turn { + error: Some(error.clone()), + ..test_turn("new-turn", TurnStatus::Failed, Vec::new()) + }; + for (stored_turn, notification) in [ + ( + previous_turn.clone(), + ServerNotification::Error(codex_app_server_protocol::ErrorNotification { + thread_id: thread_id.to_string(), + turn_id: "new-turn".into(), + error: error.clone(), + will_retry: false, + }), + ), + ( + previous_turn, + ServerNotification::TurnCompleted(TurnCompletedNotification { + thread_id: thread_id.to_string(), + turn: failed_turn.clone(), + }), + ), + // A settings-operation error has a submission ID, not evidence of a new turn. + ( + failed_turn, + ServerNotification::Error(codex_app_server_protocol::ErrorNotification { + thread_id: thread_id.to_string(), + turn_id: "settings-update".into(), + error: AppServerTurnError { + codex_error_info: Some(AppServerCodexErrorInfo::BadRequest), + ..error + }, + will_retry: false, + }), + ), + ] { + for saw_start in [false, true] { + let mut app = make_test_app().await; + let mut store = ThreadEventStore::new_with_session( + /*capacity*/ 1, + test_thread_session(thread_id, app.config.cwd.to_path_buf()), + vec![stored_turn.clone()], + ); + if saw_start { + store.push_notification(turn_started_notification(thread_id, "new-turn")); + } + store.push_notification_ref(¬ification); + assert_eq!(store.buffer.len(), 1); + app.replay_thread_snapshot(store.snapshot(), /*resume_restored_queue*/ false); + assert!(app.chat_widget.has_misalignment_policy_violation()); + assert!(render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("Chat stopped")); + } + } +} diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_failed_tool.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_failed_tool.snap new file mode 100644 index 0000000000..205a724501 --- /dev/null +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_failed_tool.snap @@ -0,0 +1,6 @@ +--- +source: tui/src/app/tests/buffered_replay.rs +expression: rendered +--- +• Called sample.inspect({}) + └ Error: interrupted diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_input.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_input.snap new file mode 100644 index 0000000000..2b4062523e --- /dev/null +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_input.snap @@ -0,0 +1,11 @@ +--- +source: tui/src/app/tests/buffered_replay.rs +expression: "normalize_snapshot_paths(render_bottom_popup(&app.chat_widget, 80))" +--- +• Queued follow-up inputs + ↳ keep queued + ⌥ + ↑ edit last queued message + +› keep draft + + gpt-5.6-sol default · /tmp/project diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index 32e2d9f368..c866c288c0 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -45,6 +45,8 @@ pub(super) struct ThreadEventStore { pub(super) buffer: VecDeque, pub(super) pending_interactive_replay: PendingInteractiveReplayState, pub(super) active_turn_id: Option, + // Lifecycle identity must survive bounded replay-buffer eviction. + pub(super) latest_turn_id: Option, pub(super) pending_interrupt_turn_id: Option, pub(super) input_state: Option, pub(super) capacity: usize, @@ -74,6 +76,7 @@ impl ThreadEventStore { buffer: VecDeque::new(), pending_interactive_replay: PendingInteractiveReplayState::default(), active_turn_id: None, + latest_turn_id: None, pending_interrupt_turn_id: None, input_state: None, capacity, @@ -113,9 +116,15 @@ impl ThreadEventStore { .rev() .find(|turn| matches!(turn.status, TurnStatus::InProgress)) .map(|turn| turn.id.clone()); + self.latest_turn_id = turns.last().map(|turn| turn.id.clone()); self.turns = turns; } + pub(super) fn set_active_turn_id(&mut self, turn_id: String) { + self.latest_turn_id = Some(turn_id.clone()); + self.active_turn_id = Some(turn_id); + } + pub(super) fn push_notification(&mut self, notification: ServerNotification) { self.push_notification_inner(Cow::Owned(notification)); } @@ -129,9 +138,12 @@ impl ThreadEventStore { .note_server_notification(notification.as_ref()); match notification.as_ref() { ServerNotification::TurnStarted(turn) => { - self.active_turn_id = Some(turn.turn.id.clone()); + self.set_active_turn_id(turn.turn.id.clone()); } ServerNotification::TurnCompleted(turn) => { + if self.active_turn_id.is_none() { + self.latest_turn_id = Some(turn.turn.id.clone()); + } if matches!(turn.turn.status, TurnStatus::Completed) { self.recap_progress.completed_turns += 1; } @@ -142,6 +154,14 @@ impl ThreadEventStore { self.pending_interrupt_turn_id = None; } } + ServerNotification::Error(n) + if self.active_turn_id.is_none() + && !n.will_retry + && n.error.codex_error_info + == Some(AppServerCodexErrorInfo::MisalignmentPolicyViolation) => + { + self.latest_turn_id = Some(n.turn_id.clone()); + } ServerNotification::ThreadClosed(_) => { self.active_turn_id = None; self.pending_interrupt_turn_id = None; @@ -207,7 +227,7 @@ impl ThreadEventStore { } pub(super) fn snapshot(&self) -> ThreadEventSnapshot { - ThreadEventSnapshot { + let mut snapshot = ThreadEventSnapshot { session: self.session.clone(), turns: self.turns.clone(), // Thread switches replay buffered events into a rebuilt ChatWidget. Only replay @@ -226,7 +246,11 @@ impl ThreadEventStore { .cloned() .collect(), input_state: self.input_state.clone(), + }; + if let Some(latest_turn_id) = &self.latest_turn_id { + replay_filter::omit_resolved_misalignment_errors(&mut snapshot, latest_turn_id); } + snapshot } pub(super) fn recap_progress(&self) -> recap::RecapProgress { diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 98b443f264..b78ea15dad 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -715,7 +715,7 @@ impl App { self.thread_event_channels.get(&thread_id) { let mut store = channel.store.lock().await; - store.active_turn_id = Some(actual_turn_id.clone()); + store.set_active_turn_id(actual_turn_id.clone()); } steer_turn_id = actual_turn_id; retried_after_turn_mismatch = true; @@ -727,7 +727,7 @@ impl App { self.thread_event_channels.get(&thread_id) { let mut store = channel.store.lock().await; - store.active_turn_id = Some(actual_turn_id); + store.set_active_turn_id(actual_turn_id); } return Err(error.into()); } @@ -803,7 +803,7 @@ impl App { .wrap_err("review/start returned invalid review thread id")?; let store = Arc::clone(&self.ensure_thread_channel(review_thread_id).store); let mut store = store.lock().await; - store.active_turn_id = Some(response.turn.id); + store.set_active_turn_id(response.turn.id); Ok(true) } AppCommand::CleanBackgroundTerminals => { diff --git a/codex-rs/tui/src/chatwidget/replay.rs b/codex-rs/tui/src/chatwidget/replay.rs index 1660b702e9..1c3f3e6381 100644 --- a/codex-rs/tui/src/chatwidget/replay.rs +++ b/codex-rs/tui/src/chatwidget/replay.rs @@ -30,6 +30,7 @@ impl ChatWidget { /// avoid triggering side effects. Event ids are passed as `None` to /// distinguish replayed events from live ones. pub(crate) fn replay_thread_turns(&mut self, turns: Vec, replay_kind: ReplayKind) { + let latest_turn_id = turns.last().map(|turn| turn.id.clone()); let hidden_nested_review_turns = std::iter::once(/*value*/ false) .chain(turns.windows(/*size*/ 2).map(|turns| { crate::app_backtrack::is_hidden_nested_review_turn(&turns[0], &turns[1]) @@ -41,7 +42,7 @@ impl ChatWidget { items_view: _, items, status, - error, + mut error, started_at, completed_at, duration_ms, @@ -62,6 +63,15 @@ impl ChatWidget { } else { status }; + // A resolved historical precaution must not clear the restored draft or input queue. + if Some(&turn_id) != latest_turn_id.as_ref() + && error.as_ref().is_some_and(|error| { + error.codex_error_info + == Some(AppServerCodexErrorInfo::MisalignmentPolicyViolation) + }) + { + error = None; + } if matches!( status, TurnStatus::Completed | TurnStatus::Interrupted | TurnStatus::Failed