openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tui): preserve streamed tail across lifecycle changes

Browse Source
This commit is contained in:
Felipe Coury
2026-06-05 15:43:34 -03:00
parent 31bd9c3fe9
commit 072b439810
5 changed files with 309 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/chatwidget.rs
+9 -2
View File
@@ -1624,11 +1624,18 @@ impl ChatWidget {
self.raw_output_mode = enabled;
self.config.tui_raw_output_mode = enabled;
let render_mode = self.history_render_mode();
let mut stream_state_changed = false;
if let Some(controller) = self.stream_controller.as_mut() {
controller.set_render_mode(render_mode);
stream_state_changed |= controller.set_render_mode(render_mode);
}
if let Some(controller) = self.plan_stream_controller.as_mut() {
controller.set_render_mode(render_mode);
stream_state_changed |= controller.set_render_mode(render_mode);
}
if stream_state_changed {
if !self.stream_controllers_idle() {
self.app_event_tx.send(AppEvent::StartCommitAnimation);
}
self.sync_active_stream_tail();
}
self.refresh_status_surfaces();
}
codex-rs/tui/src/chatwidget/tests/status_and_layout.rs
+75
View File
@@ -1409,6 +1409,81 @@ async fn streaming_final_answer_keeps_task_running_state() {
assert!(!chat.bottom_pane.quit_shortcut_hint_visible());
}
#[tokio::test]
async fn finalize_turn_persists_held_stream_tail() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.on_task_started();
chat.on_agent_message_delta("prefix already emitted\n".to_string());
chat.on_agent_message_delta("tail remains visible\n".to_string());
chat.on_commit_tick();
drain_insert_history(&mut rx);
let active_before = chat
.transcript
.active_cell
.as_ref()
.map(|cell| lines_to_single_string(&cell.display_lines(/*width*/ 80)))
.unwrap_or_default();
assert!(
active_before.contains("tail remains visible"),
"expected held tail before termination, got {active_before:?}",
);
chat.finalize_turn();
let inserted = drain_insert_history(&mut rx);
let rendered = inserted
.iter()
.map(|cell| lines_to_single_string(cell))
.collect::<Vec<_>>()
.join("\n");
assert!(
rendered.contains("tail remains visible"),
"expected held tail to be persisted on termination, got {rendered:?}",
);
assert!(chat.transcript.active_cell.is_none());
assert!(chat.stream_controller.is_none());
}
#[tokio::test]
async fn raw_output_toggle_refreshes_active_stream_tail() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.on_task_started();
chat.on_agent_message_delta("prefix already emitted\n".to_string());
chat.on_agent_message_delta("**tail remains visible**\n".to_string());
chat.on_commit_tick();
let rich_tail = chat
.transcript
.active_cell
.as_ref()
.map(|cell| lines_to_single_string(&cell.display_lines(/*width*/ 80)))
.unwrap_or_default();
assert!(
rich_tail.contains("tail remains visible"),
"expected rich tail before toggle, got {rich_tail:?}",
);
assert!(
!rich_tail.contains("**tail remains visible**"),
"expected rich tail to render markdown before toggle, got {rich_tail:?}",
);
chat.set_raw_output_mode(/*enabled*/ true);
let raw_tail = chat
.transcript
.active_cell
.as_ref()
.map(|cell| lines_to_single_string(&cell.display_lines(/*width*/ 80)))
.unwrap_or_default();
assert!(
raw_tail.contains("**tail remains visible**"),
"expected active tail to refresh after raw toggle, got {raw_tail:?}",
);
}
#[tokio::test]
async fn ctrl_c_interrupt_pauses_active_goal_turn() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
codex-rs/tui/src/chatwidget/turn_runtime.rs
+13 -5
View File
@@ -294,9 +294,19 @@ impl ChatWidget {
/// This does not clear MCP startup tracking, because MCP startup can overlap with turn cleanup
/// and should continue to drive the bottom-pane running indicator while it is in progress.
pub(super) fn finalize_turn(&mut self) {
// Drop preview-only stream tail content on any termination path before
// failed-cell finalization, so transient tail cells are never persisted.
self.clear_active_stream_tail();
self.flush_answer_stream_with_separator();
if let Some(mut controller) = self.plan_stream_controller.take() {
let requires_scrollback_reflow = controller.requires_final_scrollback_reflow();
self.clear_active_stream_tail();
let (cell, source) = controller.finalize();
if !requires_scrollback_reflow && let Some(cell) = cell {
self.add_boxed_history(cell);
}
if let Some(source) = source {
self.app_event_tx
.send(AppEvent::ConsolidateProposedPlan(source));
}
}
// Ensure any spinner is replaced by a red ✗ and flushed into history.
self.finalize_active_cell_as_failed();
// Turn-scoped hook rows are transient live state; once the turn is over,
@@ -314,8 +324,6 @@ impl ChatWidget {
self.last_unified_wait = None;
self.unified_exec_wait_streak = None;
self.adaptive_chunking.reset();
self.stream_controller = None;
self.plan_stream_controller = None;
self.status_state.pending_status_indicator_restore = false;
self.clear_cancel_edit();
self.request_status_line_branch_refresh();
codex-rs/tui/src/streaming/controller.rs
+150 -61
View File
@@ -250,6 +250,8 @@ impl StreamCore {
}
let had_pending_queue = self.state.queued_len() > 0;
let had_structural_tail = self.requires_final_scrollback_reflow();
let emitted_source_boundary =
self.source_boundary_after_rendered_len(self.emitted_stable_len);
self.width = width;
self.state.collector.set_width(width);
if self.raw_source.is_empty() {
@@ -257,16 +259,13 @@ impl StreamCore {
}
self.recompute_streaming_render();
self.emitted_stable_len = self.emitted_stable_len.min(self.rendered_lines.len());
self.emitted_stable_len = self.rendered_len_for_source_boundary(emitted_source_boundary);
let target_stable_len = self.compute_target_stable_len();
if had_pending_queue
&& self.emitted_stable_len >= target_stable_len
&& target_stable_len > 0
{
// If wrapped remainder compresses into fewer lines at the new width,
// keep at least one line un-emitted so pre-resize pending content is
// not skipped permanently.
self.emitted_stable_len = target_stable_len - 1;
self.emitted_stable_len = target_stable_len;
}
self.state.clear_queue();
if self.emitted_stable_len > 0 && !had_pending_queue && !had_structural_tail {
@@ -309,9 +308,9 @@ impl StreamCore {
self.rendered_lines = self.render_source(&self.raw_source);
}
fn set_render_mode(&mut self, render_mode: HistoryRenderMode) {
fn set_render_mode(&mut self, render_mode: HistoryRenderMode) -> bool {
if self.render_mode == render_mode {
return;
return false;
}
let had_pending_queue = self.state.queued_len() > 0;
@@ -321,20 +320,22 @@ impl StreamCore {
} else {
None
};
let emitted_source_boundary =
self.source_boundary_after_rendered_len(self.emitted_stable_len);
self.render_mode = render_mode;
self.stable_prefix_len_cache = None;
if self.raw_source.is_empty() {
return;
return false;
}
self.recompute_streaming_render();
self.emitted_stable_len = self.emitted_stable_len.min(self.rendered_lines.len());
self.emitted_stable_len = self.rendered_len_for_source_boundary(emitted_source_boundary);
let target_stable_len = self.compute_target_stable_len();
if had_pending_queue
&& self.emitted_stable_len >= target_stable_len
&& target_stable_len > 0
{
self.emitted_stable_len = target_stable_len - 1;
self.emitted_stable_len = target_stable_len;
}
self.state.clear_queue();
if self.emitted_stable_len > 0 && !had_pending_queue && !had_structural_tail {
@@ -345,9 +346,10 @@ impl StreamCore {
};
self.emitted_stable_len = target_stable_len;
self.enqueued_stable_len = target_stable_len;
return;
return true;
}
self.rebuild_stable_queue_from_render();
true
}
/// Compute how many rendered lines should be in the stable region.
@@ -487,6 +489,41 @@ impl StreamCore {
stable_prefix_len
}
fn rendered_len_for_source_boundary(&mut self, source_boundary: usize) -> usize {
if source_boundary == 0 {
return 0;
}
if source_boundary >= self.raw_source.len() {
return self.rendered_lines.len();
}
self.stable_prefix_len_for_source_start(source_boundary)
}
fn source_boundary_after_rendered_len(&self, rendered_len: usize) -> usize {
if rendered_len == 0 || self.raw_source.is_empty() {
return 0;
}
if rendered_len >= self.rendered_lines.len() {
return self.raw_source.len();
}
let mut boundaries = self
.raw_source
.match_indices('\n')
.map(|(idx, _)| idx + 1)
.collect::<Vec<_>>();
if boundaries.last().copied() != Some(self.raw_source.len()) {
boundaries.push(self.raw_source.len());
}
for boundary in boundaries {
if self.render_source(&self.raw_source[..boundary]).len() >= rendered_len {
return boundary;
}
}
self.raw_source.len()
}
fn default_tail_budget_lines(&mut self) -> usize {
if self.default_tail_lines == 0 || self.rendered_lines.is_empty() {
return 0;
@@ -631,8 +668,8 @@ impl StreamController {
self.core.set_width(width);
}
pub(crate) fn set_render_mode(&mut self, render_mode: HistoryRenderMode) {
self.core.set_render_mode(render_mode);
pub(crate) fn set_render_mode(&mut self, render_mode: HistoryRenderMode) -> bool {
self.core.set_render_mode(render_mode)
}
fn emit(&mut self, lines: Vec<HyperlinkLine>) -> Option<Box<dyn HistoryCell>> {
@@ -761,8 +798,8 @@ impl PlanStreamController {
self.core.set_width(width);
}
pub(crate) fn set_render_mode(&mut self, render_mode: HistoryRenderMode) {
self.core.set_render_mode(render_mode);
pub(crate) fn set_render_mode(&mut self, render_mode: HistoryRenderMode) -> bool {
self.core.set_render_mode(render_mode)
}
fn emit(
@@ -838,6 +875,18 @@ mod tests {
PlanStreamController::new(width, &test_cwd(), HistoryRenderMode::Rich)
}
const HAPPY_PATH_DUPLICATE_REPRO_DELTAS: &[&str] = &[
"**Happy Path**\n",
"\n",
"1. In the second terminal, make the smoke home non-writable:\n",
" ```fish\n",
" chmod a-w $CODEX_SMOKE_HOME\n",
" ```\n",
" Expected: the running TUI stays open.\n",
"\n",
"2. In the TUI, type `/model` and press Enter.\n",
];
fn lines_to_plain_strings(lines: &[ratatui::text::Line<'_>]) -> Vec<String> {
lines
.iter()
@@ -876,6 +925,38 @@ mod tests {
.collect()
}
fn collect_visible_stream_snapshots(deltas: &[&str], width: Option<usize>) -> Vec<Vec<String>> {
let mut ctrl = stream_controller(width);
let mut committed_lines = Vec::new();
let mut snapshots = Vec::new();
for delta in deltas {
ctrl.push(delta);
while let (Some(cell), idle) = ctrl.on_commit_tick() {
committed_lines.extend(cell.transcript_lines(u16::MAX));
if idle {
break;
}
}
let mut visible = lines_to_plain_strings(&committed_lines)
.into_iter()
.map(|s| s.chars().skip(2).collect::<String>())
.collect::<Vec<_>>();
visible.extend(hyperlink_lines_to_plain_strings(&ctrl.current_tail_lines()));
snapshots.push(visible);
}
snapshots
}
fn format_stream_snapshots(snapshots: &[Vec<String>]) -> String {
snapshots
.iter()
.enumerate()
.map(|(idx, lines)| format!("after delta {idx}:\n{}", lines.join("\n")))
.collect::<Vec<_>>()
.join("\n---\n")
}
fn collect_plan_streamed_lines(deltas: &[&str], width: Option<usize>) -> Vec<String> {
let mut ctrl = plan_stream_controller(width);
let mut lines = Vec::new();
@@ -1008,31 +1089,39 @@ mod tests {
#[test]
fn controller_does_not_duplicate_list_continuation_after_fenced_code() {
let lines = collect_streamed_lines(
&[
"**Happy Path**\n",
"\n",
"1. In the second terminal, make the smoke home non-writable:\n",
" ```fish\n",
" chmod a-w $CODEX_SMOKE_HOME\n",
" ```\n",
" Expected: the running TUI stays open.\n",
"\n",
"2. In the TUI, type `/model` and press Enter.\n",
],
Some(/*width*/ 80),
);
let snapshots =
collect_visible_stream_snapshots(HAPPY_PATH_DUPLICATE_REPRO_DELTAS, Some(/*width*/ 80));
let expected_count = lines
.iter()
for lines in &snapshots {
let expected_count = lines
.iter()
.filter(|line| line.trim() == "Expected: the running TUI stays open.")
.count();
assert!(
expected_count <= 1,
"expected list continuation to render at most once, got {lines:?}",
);
}
let final_count = snapshots
.last()
.into_iter()
.flatten()
.filter(|line| line.trim() == "Expected: the running TUI stays open.")
.count();
assert_eq!(
expected_count, 1,
"expected list continuation to render once, got {lines:?}",
final_count, 1,
"expected final visible snapshot to include the list continuation once"
);
}
#[test]
fn controller_streamed_list_continuation_snapshot() {
let snapshots =
collect_visible_stream_snapshots(HAPPY_PATH_DUPLICATE_REPRO_DELTAS, Some(/*width*/ 80));
insta::assert_snapshot!(format_stream_snapshots(&snapshots));
}
#[test]
fn controller_live_tail_rerenders_table_tail_after_resize() {
let mut ctrl = stream_controller(Some(96));
@@ -1088,43 +1177,36 @@ mod tests {
}
#[test]
fn controller_set_width_partial_drain_keeps_pending_queue() {
fn controller_set_width_partial_drain_does_not_duplicate_emitted_prefix() {
let mut ctrl = stream_controller(Some(40));
ctrl.push("AAAA BBBB CCCC DDDD EEEE FFFF GGGG HHHH IIII JJJJ\n");
ctrl.push("second line\n");
let (cell, idle) = ctrl.on_commit_tick();
assert!(cell.is_some(), "expected 1 emitted line");
let emitted_before_resize = cell
.expect("expected 1 emitted line")
.transcript_lines(u16::MAX);
assert!(!idle, "queue should still have lines");
assert!(ctrl.queued_lines() > 0, "expected pending queued lines");
ctrl.set_width(Some(20));
ctrl.set_width(Some(/*width*/ 120));
assert!(
ctrl.queued_lines() > 0,
"resize must preserve pending queued lines"
);
let mut drained = Vec::new();
for _ in 0..64 {
let (cell, is_idle) = ctrl.on_commit_tick();
if let Some(cell) = cell {
drained.extend(lines_to_plain_strings(&cell.transcript_lines(u16::MAX)));
}
if is_idle {
break;
}
}
assert!(
drained.iter().any(|l| l.contains("IIII JJJJ")),
"pending wrapped lines should continue draining after resize; got {drained:?}",
assert_eq!(
ctrl.queued_lines(),
0,
"partially emitted source line should not be re-queued after widening"
);
let (cell, _source) = ctrl.finalize();
let final_lines = cell
.map(|cell| lines_to_plain_strings(&cell.transcript_lines(u16::MAX)))
.unwrap_or_default();
let emitted_before_resize = lines_to_plain_strings(&emitted_before_resize).join("\n");
let finalized_after_resize = final_lines.join("\n");
assert!(
!finalized_after_resize.contains("AAAA BBBB CCCC"),
"finalize must not replay the already-visible line prefix; emitted before resize: {emitted_before_resize:?}, finalized after resize: {finalized_after_resize:?}",
);
assert!(
final_lines.iter().any(|line| line.contains("second line")),
"ordinary mutable tail should flush on finalize; got {final_lines:?}",
@@ -2106,7 +2188,7 @@ mod tests {
"expected non-empty queue before resize"
);
ctrl.set_width(Some(120));
ctrl.set_width(Some(/*width*/ 120));
let (cell, _source) = ctrl.finalize();
let remaining = cell
@@ -2122,13 +2204,15 @@ mod tests {
}
#[test]
fn controller_set_width_partial_wrapped_emit_keeps_wrapped_remainder() {
fn controller_set_width_partial_wrapped_emit_does_not_replay_prefix() {
let mut ctrl = stream_controller(Some(18));
ctrl.push("alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu\n");
ctrl.push("tail line\n");
let (first_emit, idle) = ctrl.on_commit_tick();
assert!(first_emit.is_some(), "expected first wrapped line emission");
let first_emit = first_emit
.expect("expected first wrapped line emission")
.transcript_lines(u16::MAX);
assert!(!idle, "expected remaining wrapped content after one tick");
assert!(
ctrl.queued_lines() > 0,
@@ -2141,10 +2225,15 @@ mod tests {
let remaining = cell
.map(|c| lines_to_plain_strings(&c.transcript_lines(u16::MAX)))
.unwrap_or_default();
let joined = remaining.join(" ");
let first_emit = lines_to_plain_strings(&first_emit).join("\n");
let joined = remaining.join("\n");
assert!(
joined.contains("kappa") || joined.contains("lambda") || joined.contains("mu"),
"wrapped remainder from partially emitted source line was lost after resize: {remaining:?}",
!joined.contains("alpha beta"),
"finalize must not replay the already-visible prefix; emitted before resize: {first_emit:?}, finalized after resize: {joined:?}",
);
assert!(
remaining.iter().any(|line| line.contains("tail line")),
"ordinary mutable tail should flush on finalize; got {remaining:?}",
);
}
}
codex-rs/tui/src/streaming/snapshots/codex_tui__streaming__controller__tests__controller_streamed_list_continuation_snapshot.snap
+62
View File
@@ -0,0 +1,62 @@
---
source: tui/src/streaming/controller.rs
assertion_line: 1122
expression: format_stream_snapshots(&snapshots)
---
after delta 0:
Happy Path
---
after delta 1:
Happy Path
---
after delta 2:
Happy Path
1. In the second terminal, make the smoke home non-writable:
---
after delta 3:
Happy Path
1. In the second terminal, make the smoke home non-writable:
---
after delta 4:
Happy Path
1. In the second terminal, make the smoke home non-writable:
chmod a-w $CODEX_SMOKE_HOME
---
after delta 5:
Happy Path
1. In the second terminal, make the smoke home non-writable:
chmod a-w $CODEX_SMOKE_HOME
---
after delta 6:
Happy Path
1. In the second terminal, make the smoke home non-writable:
chmod a-w $CODEX_SMOKE_HOME
Expected: the running TUI stays open.
---
after delta 7:
Happy Path
1. In the second terminal, make the smoke home non-writable:
chmod a-w $CODEX_SMOKE_HOME
Expected: the running TUI stays open.
---
after delta 8:
Happy Path
1. In the second terminal, make the smoke home non-writable:
chmod a-w $CODEX_SMOKE_HOME
Expected: the running TUI stays open.
2. In the TUI, type /model and press Enter.