github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 16:38:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve restored input after resolved misalignment errors (#42399)

Browse Source 
## 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
This commit is contained in:
Eric Traut
2026-09-03 00:36:29 +00:00
committed by copyberry
parent e6a944ad75
commit 54a4077c8b
7 changed files with 266 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/replay_filter.rs
+31
View File
@@ -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.
codex-rs/tui/src/app/tests/buffered_replay.rs
+178
View File
@@ -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::<String>();
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(&notification);
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"));
}
}
}
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_failed_tool.snap
+6
View File
@@ -0,0 +1,6 @@
---
source: tui/src/app/tests/buffered_replay.rs
expression: rendered
---
• Called sample.inspect({})
└ Error: interrupted
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__buffered_replay__misalignment_replayed_input.snap
+11
View File
@@ -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
codex-rs/tui/src/app/thread_events.rs
+26 -2
View File
@@ -45,6 +45,8 @@ pub(super) struct ThreadEventStore {
pub(super) buffer: VecDeque<ThreadBufferedEvent>,
pub(super) pending_interactive_replay: PendingInteractiveReplayState,
pub(super) active_turn_id: Option<String>,
// Lifecycle identity must survive bounded replay-buffer eviction.
pub(super) latest_turn_id: Option<String>,
pub(super) pending_interrupt_turn_id: Option<String>,
pub(super) input_state: Option<ThreadInputState>,
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 {
codex-rs/tui/src/app/thread_routing.rs
+3 -3
View File
@@ -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 => {
codex-rs/tui/src/chatwidget/replay.rs
+11 -1
View File
@@ -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<Turn>, 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