From 7f75cd211d4ccb38f2fda00e5ab8998dcb3403d2 Mon Sep 17 00:00:00 2001 From: Felipe Coury Date: Sun, 5 Jul 2026 19:57:26 -0300 Subject: [PATCH] feat(tui): retain owned-screen stream source --- codex-rs/tui/src/app/conversation_panes.rs | 9 + codex-rs/tui/src/app/event_dispatch.rs | 9 +- codex-rs/tui/src/app/session_lifecycle.rs | 6 + codex-rs/tui/src/app/tests.rs | 4 +- codex-rs/tui/src/app_event.rs | 6 + codex-rs/tui/src/chatwidget.rs | 14 +- codex-rs/tui/src/chatwidget/constructor.rs | 3 +- codex-rs/tui/src/chatwidget/streaming.rs | 153 ++++++++-- .../src/chatwidget/tests/slash_commands.rs | 7 +- ...ut__retained_live_agent_table_reflows.snap | 19 ++ .../src/chatwidget/tests/status_and_layout.rs | 181 +++++++++++- codex-rs/tui/src/chatwidget/tokens.rs | 24 +- codex-rs/tui/src/chatwidget/transcript.rs | 2 + codex-rs/tui/src/chatwidget/turn_runtime.rs | 51 +++- codex-rs/tui/src/session_log.rs | 9 + codex-rs/tui/src/streaming/controller.rs | 274 +++++++++++++++--- codex-rs/tui/src/streaming/mod.rs | 3 + codex-rs/tui/src/streaming/surface.rs | 21 ++ 18 files changed, 696 insertions(+), 99 deletions(-) create mode 100644 codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__status_and_layout__retained_live_agent_table_reflows.snap create mode 100644 codex-rs/tui/src/streaming/surface.rs diff --git a/codex-rs/tui/src/app/conversation_panes.rs b/codex-rs/tui/src/app/conversation_panes.rs index 6ffff42ea5..e4ab00d7b9 100644 --- a/codex-rs/tui/src/app/conversation_panes.rs +++ b/codex-rs/tui/src/app/conversation_panes.rs @@ -27,6 +27,7 @@ use crate::file_search::FileSearchManager; use crate::history_cell::HistoryCell; use crate::key_hint; use crate::key_hint::KeyBinding; +use crate::streaming::StreamSurface; use crate::transcript_reflow::TranscriptReflowState; use crate::tui::MousePrimaryEvent; @@ -67,6 +68,14 @@ impl ConversationPane { return Err(init); } + let mut init = init; + let stream_surface = if init.owned_screen.is_some() { + StreamSurface::Retained + } else { + StreamSurface::Inline + }; + init.chat_widget.set_stream_surface(stream_surface); + Ok(Self { slot, chat_widget: init.chat_widget, diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index a5a5558ea6..a9c3f6f033 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -279,6 +279,11 @@ impl App { AppEvent::InsertHistoryCell(cell) => { self.insert_history_cell(tui, cell); } + AppEvent::CommitRetainedStreamCell(cell) => { + self.insert_history_cell(tui, cell); + self.chat_widget.note_stream_commit_completed(); + self.insert_pending_usage_output_after_stream_shutdown(tui); + } AppEvent::EndInitialHistoryReplayBuffer => { self.finish_initial_history_replay_buffer(tui); } @@ -295,7 +300,7 @@ impl App { scrollback_reflow, deferred_history_cell, )?; - self.chat_widget.note_stream_consolidation_completed(); + self.chat_widget.note_stream_commit_completed(); self.insert_pending_usage_output_after_stream_shutdown(tui); } AppEvent::ConsolidateProposedPlan(source) => { @@ -334,7 +339,7 @@ impl App { self.maybe_finish_stream_reflow(tui)?; } - self.chat_widget.note_stream_consolidation_completed(); + self.chat_widget.note_stream_commit_completed(); self.insert_pending_usage_output_after_stream_shutdown(tui); } AppEvent::ApplyThreadRollback { num_turns } => { diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index cd987bcb79..0b16d0a666 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -350,6 +350,12 @@ impl App { chat_widget.last_terminal_title = previous_terminal_title; } chat_widget.remote_connection = self.chat_widget.remote_connection.clone(); + let stream_surface = if self.chat_widget.owned_screen.is_some() { + crate::streaming::StreamSurface::Retained + } else { + crate::streaming::StreamSurface::Inline + }; + chat_widget.set_stream_surface(stream_surface); for (thread_id, entry) in self.agent_navigation.ordered_threads() { chat_widget.set_collab_agent_metadata( thread_id, diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 0a9f5fc75e..3c70bac933 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -6096,14 +6096,14 @@ async fn late_usage_result_can_follow_finalized_plan() { other => panic!("expected token activity refresh request, got {other:?}"), }; - app.chat_widget.note_stream_consolidation_queued(); + app.chat_widget.note_stream_commit_queued(); app.chat_widget .transcript_cells .push(Arc::new(history_cell::new_proposed_plan_stream( vec![Line::from("finalized plan")], /*is_stream_continuation*/ false, ))); - app.chat_widget.note_stream_consolidation_completed(); + app.chat_widget.note_stream_commit_completed(); assert!( app.chat_widget.finish_token_activity_refresh( diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 495899d5d5..bda71663c0 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -728,6 +728,12 @@ pub(crate) enum AppEvent { InsertHistoryCell(Box), + /// Atomically commit the final semantic cell for a retained stream. + /// + /// The app releases the stream-ordering barrier only after inserting this cell, preventing + /// asynchronous usage output from overtaking the finalized assistant message or plan. + CommitRetainedStreamCell(Box), + /// Finish buffering initial resume replay after all replay events have been queued. EndInitialHistoryReplayBuffer, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 473cc9cecb..fd5c2d75a4 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -448,6 +448,7 @@ use self::user_messages::user_message_preview_text; mod warnings; use self::warnings::WarningDisplayState; pub(crate) use crate::branch_summary::StatusLineGitSummary; +use crate::streaming::StreamSurface; use crate::streaming::chunking::AdaptiveChunkingPolicy; use crate::streaming::commit_tick::CommitTickScope; use crate::streaming::commit_tick::run_commit_tick; @@ -572,11 +573,13 @@ pub(crate) struct ChatWidget { rate_limit_switch_prompt: RateLimitSwitchPromptState, add_credits_nudge_email_in_flight: Option, adaptive_chunking: AdaptiveChunkingPolicy, + /// Selects whether active streams emit immutable terminal rows or one retained live entry. + stream_surface: StreamSurface, // Stream lifecycle controller stream_controller: Option, // Stream lifecycle controller for proposed plan output. plan_stream_controller: Option, - pending_stream_consolidations: usize, + pending_stream_commits: usize, /// Holds the platform clipboard lease so copied text remains available while supported. clipboard_lease: Option, copy_last_response_binding: Vec, @@ -1220,6 +1223,7 @@ impl ChatWidget { } fn flush_active_cell(&mut self) { + self.transcript.retained_stream_source_revision = None; if let Some(active) = self.transcript.active_cell.take() { self.transcript.needs_final_message_separator = true; self.app_event_tx.send(AppEvent::InsertHistoryCell(active)); @@ -1680,6 +1684,14 @@ impl ChatWidget { } } + pub(crate) fn set_stream_surface(&mut self, stream_surface: StreamSurface) { + debug_assert!( + self.stream_controller.is_none() && self.plan_stream_controller.is_none(), + "stream surface must be selected before streaming starts", + ); + self.stream_surface = stream_surface; + } + pub(crate) fn set_raw_output_mode(&mut self, enabled: bool) { self.raw_output_mode = enabled; self.config.tui_raw_output_mode = enabled; diff --git a/codex-rs/tui/src/chatwidget/constructor.rs b/codex-rs/tui/src/chatwidget/constructor.rs index 113f2deaf2..6a84fa8175 100644 --- a/codex-rs/tui/src/chatwidget/constructor.rs +++ b/codex-rs/tui/src/chatwidget/constructor.rs @@ -159,9 +159,10 @@ impl ChatWidget { rate_limit_switch_prompt: RateLimitSwitchPromptState::default(), add_credits_nudge_email_in_flight: None, adaptive_chunking: AdaptiveChunkingPolicy::default(), + stream_surface: StreamSurface::Inline, stream_controller: None, plan_stream_controller: None, - pending_stream_consolidations: 0, + pending_stream_commits: 0, clipboard_lease: None, copy_last_response_binding, running_commands: HashMap::new(), diff --git a/codex-rs/tui/src/chatwidget/streaming.rs b/codex-rs/tui/src/chatwidget/streaming.rs index 9dac0c73a5..9c41de996f 100644 --- a/codex-rs/tui/src/chatwidget/streaming.rs +++ b/codex-rs/tui/src/chatwidget/streaming.rs @@ -19,15 +19,28 @@ impl ChatWidget { pub(super) fn flush_answer_stream_with_separator(&mut self) { let had_stream_controller = self.stream_controller.is_some(); if let Some(mut controller) = self.stream_controller.take() { - let scrollback_reflow = if controller.has_live_tail() { - crate::app_event::ConsolidationScrollbackReflow::Required - } else { - crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan - }; + let retained_stream = self.stream_surface == StreamSurface::Retained; + let had_live_tail = controller.has_live_tail(); self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - let deferred_history_cell = - if scrollback_reflow == crate::app_event::ConsolidationScrollbackReflow::Required { + if retained_stream { + if let Some(source) = source { + let source = parse_assistant_markdown(&source, self.config.cwd.as_path()) + .visible_markdown; + let cwd = self.config.cwd.to_path_buf(); + self.queue_retained_stream_cell(history_cell::AgentMarkdownCell::new( + source, &cwd, + )); + } + } else { + let scrollback_reflow = if had_live_tail { + crate::app_event::ConsolidationScrollbackReflow::Required + } else { + crate::app_event::ConsolidationScrollbackReflow::IfResizeReflowRan + }; + let deferred_history_cell = if scrollback_reflow + == crate::app_event::ConsolidationScrollbackReflow::Required + { cell } else { if let Some(cell) = cell { @@ -35,18 +48,19 @@ impl ChatWidget { } None }; - // Consolidate the run of streaming AgentMessageCells into a single AgentMarkdownCell - // that can re-render from source on resize. - if let Some(source) = source { - let source = - parse_assistant_markdown(&source, self.config.cwd.as_path()).visible_markdown; - self.note_stream_consolidation_queued(); - self.app_event_tx.send(AppEvent::ConsolidateAgentMessage { - source, - cwd: self.config.cwd.to_path_buf(), - scrollback_reflow, - deferred_history_cell, - }); + // Consolidate the run of streaming AgentMessageCells into a single AgentMarkdownCell + // that can re-render from source on resize. + if let Some(source) = source { + let source = parse_assistant_markdown(&source, self.config.cwd.as_path()) + .visible_markdown; + self.note_stream_commit_queued(); + self.app_event_tx.send(AppEvent::ConsolidateAgentMessage { + source, + cwd: self.config.cwd.to_path_buf(), + scrollback_reflow, + deferred_history_cell, + }); + } } } self.adaptive_chunking.reset(); @@ -132,6 +146,7 @@ impl ChatWidget { self.current_stream_width(/*reserved_cols*/ 4), &self.config.cwd, self.history_render_mode(), + self.stream_surface, )); } if let Some(controller) = self.plan_stream_controller.as_mut() @@ -174,19 +189,31 @@ impl ChatWidget { } else { (None, None) }; - if let Some(cell) = finalized_streamed_cell { + let retained_stream = + should_restore_after_stream && self.stream_surface == StreamSurface::Retained; + if retained_stream { + let source = if plan_text.is_empty() { + consolidated_plan_source.unwrap_or_default() + } else { + plan_text + }; + if !source.is_empty() { + let cwd = self.config.cwd.to_path_buf(); + self.queue_retained_stream_cell(history_cell::new_proposed_plan(source, &cwd)); + } + } else if let Some(cell) = finalized_streamed_cell { self.add_boxed_history(cell); // TODO: Replace streamed output with the final plan item text if plan streaming is // removed or if we need to reconcile mismatches between streamed and final content. if let Some(source) = consolidated_plan_source { - self.note_stream_consolidation_queued(); + self.note_stream_commit_queued(); self.app_event_tx .send(AppEvent::ConsolidateProposedPlan(source)); } } else if !plan_text.is_empty() { self.add_to_history(history_cell::new_proposed_plan(plan_text, &self.config.cwd)); } else if let Some(source) = consolidated_plan_source { - self.note_stream_consolidation_queued(); + self.note_stream_commit_queued(); self.app_event_tx .send(AppEvent::ConsolidateProposedPlan(source)); } @@ -409,6 +436,7 @@ impl ChatWidget { self.current_stream_width(/*reserved_cols*/ 2), &self.config.cwd, self.history_render_mode(), + self.stream_surface, )); } if let Some(controller) = self.stream_controller.as_mut() @@ -422,10 +450,19 @@ impl ChatWidget { } pub(super) fn active_cell_is_stream_tail(&self) -> bool { - self.transcript.active_cell.as_ref().is_some_and(|cell| { - cell.as_any().is::() - || cell.as_any().is::() - }) + self.transcript + .active_cell + .as_ref() + .is_some_and(|cell| match self.stream_surface { + StreamSurface::Inline => { + cell.as_any().is::() + || cell.as_any().is::() + } + StreamSurface::Retained => { + cell.as_any().is::() + || cell.as_any().is::() + } + }) } pub(super) fn has_active_stream_tail(&self) -> bool { @@ -435,6 +472,34 @@ impl ChatWidget { pub(super) fn sync_active_stream_tail(&mut self) { if let Some(controller) = self.stream_controller.as_ref() { + if self.stream_surface == StreamSurface::Retained { + let Some(source_revision) = controller.retained_source_revision() else { + self.clear_active_stream_tail(); + return; + }; + let active_cell_matches = self + .transcript + .active_cell + .as_ref() + .is_some_and(|cell| cell.as_any().is::()); + if self.transcript.retained_stream_source_revision == Some(source_revision) + && active_cell_matches + { + return; + } + let Some(source) = controller.retained_live_source().map(str::to_owned) else { + self.clear_active_stream_tail(); + return; + }; + self.bottom_pane.hide_status_indicator(); + self.transcript.active_cell = Some(Box::new(history_cell::AgentMarkdownCell::new( + source, + &self.config.cwd, + ))); + self.transcript.retained_stream_source_revision = Some(source_revision); + self.bump_active_cell_revision(); + return; + } let tail_lines = controller.current_tail_lines(); if tail_lines.is_empty() { self.clear_active_stream_tail(); @@ -454,6 +519,34 @@ impl ChatWidget { } if let Some(controller) = self.plan_stream_controller.as_ref() { + if self.stream_surface == StreamSurface::Retained { + let Some(source_revision) = controller.retained_source_revision() else { + self.clear_active_stream_tail(); + return; + }; + let active_cell_matches = self + .transcript + .active_cell + .as_ref() + .is_some_and(|cell| cell.as_any().is::()); + if self.transcript.retained_stream_source_revision == Some(source_revision) + && active_cell_matches + { + return; + } + let Some(source) = controller.retained_live_source().map(str::to_owned) else { + self.clear_active_stream_tail(); + return; + }; + self.bottom_pane.hide_status_indicator(); + self.transcript.active_cell = Some(Box::new(history_cell::new_proposed_plan( + source, + &self.config.cwd, + ))); + self.transcript.retained_stream_source_revision = Some(source_revision); + self.bump_active_cell_revision(); + return; + } let Some(tail) = controller.current_tail_display() else { self.clear_active_stream_tail(); return; @@ -476,9 +569,17 @@ impl ChatWidget { } pub(super) fn clear_active_stream_tail(&mut self) { + self.transcript.retained_stream_source_revision = None; if self.active_cell_is_stream_tail() { self.transcript.active_cell = None; self.bump_active_cell_revision(); } } + + pub(super) fn queue_retained_stream_cell(&mut self, cell: impl HistoryCell + 'static) { + self.transcript.needs_final_message_separator = true; + self.note_stream_commit_queued(); + self.app_event_tx + .send(AppEvent::CommitRetainedStreamCell(Box::new(cell))); + } } diff --git a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs index c7cfc219c9..edbaa23044 100644 --- a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs +++ b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs @@ -1438,14 +1438,14 @@ async fn completed_token_activity_refresh_waits_for_active_stream() { } #[tokio::test] -async fn completed_token_activity_refresh_waits_for_queued_stream_consolidation() { +async fn completed_token_activity_refresh_waits_for_queued_stream_commit() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; set_chatgpt_auth(&mut chat); let request_id = dispatch_usage_and_expect_refresh(&mut chat, &mut rx); chat.on_agent_message_delta("partial response".to_string()); chat.finalize_completed_assistant_message(/*message*/ None); - assert!(chat.pending_stream_consolidations > 0); + assert!(chat.pending_stream_commits > 0); assert!( chat.finish_token_activity_refresh( @@ -1455,7 +1455,7 @@ async fn completed_token_activity_refresh_waits_for_queued_stream_consolidation( ); assert!(chat.usage_history_insertion_blocked()); - chat.note_stream_consolidation_completed(); + chat.note_stream_commit_completed(); assert!(!chat.usage_history_insertion_blocked()); } @@ -1535,6 +1535,7 @@ async fn completed_token_activity_refresh_retries_after_plan_item_completion() { /*width*/ None, &chat.config.cwd, chat.history_render_mode(), + crate::streaming::StreamSurface::Inline, ); controller.push("Plan details"); chat.plan_stream_controller = Some(controller); diff --git a/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__status_and_layout__retained_live_agent_table_reflows.snap b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__status_and_layout__retained_live_agent_table_reflows.snap new file mode 100644 index 0000000000..8082407d23 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__status_and_layout__retained_live_agent_table_reflows.snap @@ -0,0 +1,19 @@ +--- +source: tui/src/chatwidget/tests/status_and_layout.rs +expression: "format!(\"wide (72 columns):\\n{wide}\\nnarrow (32 columns):\\n{narrow}\")" +--- +wide (72 columns): +• Table status: + + Item Notes + ━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ + alpha a longer description that wraps when narrow + +narrow (32 columns): +• Table status: + + Item Notes + ━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━ + alpha a longer + description that + wraps when narrow diff --git a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs index c9596a69e1..6815b895ed 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -258,15 +258,188 @@ async fn raw_output_mode_can_change_without_inserting_notice() { ); } +fn install_retained_plan_stream(chat: &mut ChatWidget, source: &str) { + let mut plan = crate::streaming::controller::PlanStreamController::new( + /*width*/ Some(80), + &chat.config.cwd, + HistoryRenderMode::Rich, + crate::streaming::StreamSurface::Retained, + ); + assert!(!plan.push(source)); + chat.plan_stream_controller = Some(plan); + chat.sync_active_stream_tail(); +} + +fn take_retained_commit( + rx: &mut tokio::sync::mpsc::UnboundedReceiver, +) -> Box { + std::iter::from_fn(|| rx.try_recv().ok()) + .find_map(|event| match event { + AppEvent::CommitRetainedStreamCell(cell) => Some(cell), + _ => None, + }) + .expect("retained stream commit") +} + +#[tokio::test] +async fn retained_stream_cells_only_rebuild_when_source_revision_changes() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_stream_surface(crate::streaming::StreamSurface::Retained); + chat.handle_streaming_delta("first line\n".to_string()); + + let first_active_revision = chat.transcript.active_cell_revision; + assert_eq!(chat.transcript.retained_stream_source_revision, Some(1)); + chat.sync_active_stream_tail(); + { + let controller = chat.stream_controller.as_mut().expect("agent controller"); + controller.set_width(/*width*/ Some(24)); + assert!(!controller.push("partial")); + } + chat.sync_active_stream_tail(); + assert_eq!( + ( + chat.transcript.retained_stream_source_revision, + chat.transcript.active_cell_revision, + ), + (Some(1), first_active_revision), + ); + + let controller = chat.stream_controller.as_mut().expect("agent controller"); + assert!(!controller.push(" second line\n")); + chat.sync_active_stream_tail(); + assert_eq!( + ( + chat.transcript.retained_stream_source_revision, + chat.transcript.active_cell_revision, + ), + (Some(2), first_active_revision + 1), + ); + + chat.clear_active_stream_tail(); + assert_eq!(chat.transcript.retained_stream_source_revision, None); +} + +#[tokio::test] +async fn retained_agent_interruption_commits_only_completed_source() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_stream_surface(crate::streaming::StreamSurface::Retained); + while rx.try_recv().is_ok() {} + + chat.handle_streaming_delta("complete line\npartial".to_string()); + chat.finalize_turn(); + + let committed = take_retained_commit(&mut rx); + assert!(committed.as_any().is::()); + assert_eq!( + lines_to_single_string(&committed.raw_lines()), + "complete line\n" + ); +} + +#[tokio::test] +async fn retained_plan_interruption_commits_only_completed_source() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_stream_surface(crate::streaming::StreamSurface::Retained); + install_retained_plan_stream(&mut chat, "1. completed step\npartial"); + while rx.try_recv().is_ok() {} + + chat.finalize_turn(); + + let committed = take_retained_commit(&mut rx); + assert!(committed.as_any().is::()); + assert_eq!( + lines_to_single_string(&committed.raw_lines()), + "1. completed step\n" + ); +} + +#[tokio::test] +async fn task_start_clears_stale_retained_plan_before_next_agent_revision() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_stream_surface(crate::streaming::StreamSurface::Retained); + install_retained_plan_stream(&mut chat, "stale plan\n"); + assert_eq!(chat.transcript.retained_stream_source_revision, Some(1)); + + chat.on_task_started(); + + assert!(chat.plan_stream_controller.is_none() && chat.transcript.active_cell.is_none()); + assert_eq!(chat.transcript.retained_stream_source_revision, None); + + chat.handle_streaming_delta("fresh answer\n".to_string()); + assert_eq!(chat.transcript.retained_stream_source_revision, Some(1)); + let active = chat + .transcript + .active_cell + .as_ref() + .expect("fresh agent cell"); + assert_eq!( + lines_to_single_string(&active.raw_lines()), + "fresh answer\n" + ); +} + +#[tokio::test] +async fn retained_live_agent_table_reflows_and_matches_final_commit() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_stream_surface(crate::streaming::StreamSurface::Retained); + while rx.try_recv().is_ok() {} + + chat.handle_streaming_delta( + "Table status:\n\n| Item | Notes |\n| --- | --- |\n| alpha | a longer description that wraps when narrow |\n" + .to_string(), + ); + let widths = [72, 32]; + let live_rendered = { + let active = chat + .transcript + .active_cell + .as_ref() + .expect("live table cell"); + widths.map(|width| lines_to_single_string(&active.display_lines(width))) + }; + + chat.flush_answer_stream_with_separator(); + + let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::>(); + assert!( + !events.iter().any(|event| matches!( + event, + AppEvent::InsertHistoryCell(_) + | AppEvent::ConsolidateAgentMessage { .. } + | AppEvent::ConsolidateProposedPlan(_) + | AppEvent::StartCommitAnimation + )), + "retained stream emitted a provisional event: {events:?}", + ); + let mut commits = events.into_iter().filter_map(|event| match event { + AppEvent::CommitRetainedStreamCell(cell) => Some(cell), + _ => None, + }); + let committed = commits.next().expect("retained stream commit"); + assert!( + commits.next().is_none(), + "expected exactly one retained commit" + ); + let final_rendered = + widths.map(|width| lines_to_single_string(&committed.display_lines(width))); + assert_eq!(final_rendered, live_rendered); + let [wide, narrow] = live_rendered; + insta::assert_snapshot!( + "retained_live_agent_table_reflows", + format!("wide (72 columns):\n{wide}\nnarrow (32 columns):\n{narrow}") + ); +} + #[tokio::test] async fn flush_answer_stream_keeps_default_reflow_for_plain_text_tail() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; let cwd = chat.config.cwd.to_path_buf(); let mut controller = crate::streaming::controller::StreamController::new( - Some(80), + /*width*/ Some(80), cwd.as_path(), HistoryRenderMode::Rich, + crate::streaming::StreamSurface::Inline, ); assert!(controller.push("plain response line\n")); chat.stream_controller = Some(controller); @@ -312,9 +485,10 @@ async fn flush_answer_stream_requests_scrollback_reflow_for_live_table_tail() { let cwd = chat.config.cwd.to_path_buf(); let mut controller = crate::streaming::controller::StreamController::new( - Some(80), + /*width*/ Some(80), cwd.as_path(), HistoryRenderMode::Rich, + crate::streaming::StreamSurface::Inline, ); controller.push("| Name | Notes |\n"); controller.push("| --- | --- |\n"); @@ -369,9 +543,10 @@ async fn completed_plan_table_tail_skips_provisional_history_insert() { let cwd = chat.config.cwd.to_path_buf(); let mut controller = crate::streaming::controller::PlanStreamController::new( - Some(80), + /*width*/ Some(80), cwd.as_path(), HistoryRenderMode::Rich, + crate::streaming::StreamSurface::Inline, ); controller.push("| Step | Owner |\n"); controller.push("| --- | --- |\n"); diff --git a/codex-rs/tui/src/chatwidget/tokens.rs b/codex-rs/tui/src/chatwidget/tokens.rs index 988b1fe2dd..f568688aed 100644 --- a/codex-rs/tui/src/chatwidget/tokens.rs +++ b/codex-rs/tui/src/chatwidget/tokens.rs @@ -7,7 +7,7 @@ //! matching response arrives, [`TokenActivityHandle`] updates the shared card state //! and [`ChatWidget::finish_token_activity_refresh`] moves the cell into a completed //! slot. Event dispatch commits that completed cell into history only after active -//! output and stream consolidation no longer block insertion. +//! output and stream commits no longer block insertion. //! //! Pure chart rendering and date bucketing live in [`chart`]. This module owns //! request correlation, transient/completed card state, and integration with @@ -232,33 +232,31 @@ impl ChatWidget { /// Reports whether completed asynchronous usage output must wait before insertion. /// - /// Inserting while a stream, queued consolidation, or active transcript cell is + /// Inserting while a stream, queued commit, or active transcript cell is /// present can reorder output relative to visible work, so callers retry once /// these barriers clear. pub(crate) fn usage_history_insertion_blocked(&self) -> bool { self.stream_controller.is_some() || self.plan_stream_controller.is_some() - || self.pending_stream_consolidations > 0 + || self.pending_stream_commits > 0 || self.transcript.active_cell.is_some() || self.active_hook_cell.is_some() } - /// Records a stream consolidation barrier that delays token card insertion. + /// Records a stream commit barrier that delays token card insertion. /// - /// Each queued consolidation should eventually call - /// [`ChatWidget::note_stream_consolidation_completed`]. - pub(crate) fn note_stream_consolidation_queued(&mut self) { - self.pending_stream_consolidations = - self.pending_stream_consolidations.saturating_add(/*rhs*/ 1); + /// Each queued commit should eventually call + /// [`ChatWidget::note_stream_commit_completed`]. + pub(crate) fn note_stream_commit_queued(&mut self) { + self.pending_stream_commits = self.pending_stream_commits.saturating_add(/*rhs*/ 1); } - /// Releases one queued stream consolidation barrier. + /// Releases one queued stream commit barrier. /// /// The counter saturates at zero so an unmatched completion does not underflow, /// but paired queue/completion calls are still the intended contract. - pub(crate) fn note_stream_consolidation_completed(&mut self) { - self.pending_stream_consolidations = - self.pending_stream_consolidations.saturating_sub(/*rhs*/ 1); + pub(crate) fn note_stream_commit_completed(&mut self) { + self.pending_stream_commits = self.pending_stream_commits.saturating_sub(/*rhs*/ 1); } /// Transfers the completed token activity card into the history insertion path. diff --git a/codex-rs/tui/src/chatwidget/transcript.rs b/codex-rs/tui/src/chatwidget/transcript.rs index 22b046728b..34c17fec6f 100644 --- a/codex-rs/tui/src/chatwidget/transcript.rs +++ b/codex-rs/tui/src/chatwidget/transcript.rs @@ -14,6 +14,8 @@ pub(super) struct TranscriptState { pub(super) active_cell: Option>, /// Monotonic-ish counter used to invalidate transcript overlay caching. pub(super) active_cell_revision: u64, + /// Source revision represented by the current retained streaming cell. + pub(super) retained_stream_source_revision: Option, /// Raw markdown of the most recently completed agent response that /// survived any local thread rollback. pub(super) last_agent_markdown: Option, diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index 6e550778a8..8abc218e51 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -59,6 +59,7 @@ impl ChatWidget { self.transcript.reset_turn_flags(); self.adaptive_chunking.reset(); if self.plan_stream_controller.take().is_some() { + self.clear_active_stream_tail(); self.request_pending_usage_output_insertion_after_stream_shutdown(); } self.turn_runtime_metrics = RuntimeMetricsSummary::default(); @@ -124,16 +125,24 @@ impl ChatWidget { // If a stream is currently active, finalize it. self.flush_answer_stream_with_separator(); if let Some(mut controller) = self.plan_stream_controller.take() { + let retained_stream = self.stream_surface == StreamSurface::Retained; let had_live_tail = controller.has_live_tail(); self.clear_active_stream_tail(); let (cell, source) = controller.finalize(); - if !had_live_tail && let Some(cell) = cell { - self.add_boxed_history(cell); - } - if let Some(source) = source { - self.note_stream_consolidation_queued(); - self.app_event_tx - .send(AppEvent::ConsolidateProposedPlan(source)); + if retained_stream { + if let Some(source) = source { + let cwd = self.config.cwd.to_path_buf(); + self.queue_retained_stream_cell(history_cell::new_proposed_plan(source, &cwd)); + } + } else { + if !had_live_tail && let Some(cell) = cell { + self.add_boxed_history(cell); + } + if let Some(source) = source { + self.note_stream_commit_queued(); + self.app_event_tx + .send(AppEvent::ConsolidateProposedPlan(source)); + } } self.request_pending_usage_output_insertion_after_stream_shutdown(); } @@ -306,9 +315,33 @@ impl ChatWidget { /// and should continue to drive the bottom-pane running indicator while it is in progress. pub(super) fn finalize_turn(&mut self) { self.clear_safety_buffering(); - // Drop preview-only stream tail content on any termination path before - // failed-cell finalization, so transient tail cells are never persisted. + let retained_sources = if self.stream_surface == StreamSurface::Retained { + ( + self.stream_controller + .as_ref() + .and_then(StreamController::retained_live_source) + .map(str::to_owned), + self.plan_stream_controller + .as_ref() + .and_then(PlanStreamController::retained_live_source) + .map(str::to_owned), + ) + } else { + (None, None) + }; + // Persist completed retained source below while dropping any incomplete + // preview suffix before failed-cell finalization. self.clear_active_stream_tail(); + if self.stream_surface == StreamSurface::Retained { + let cwd = self.config.cwd.to_path_buf(); + if let Some(source) = retained_sources.0 { + let source = parse_assistant_markdown(&source, cwd.as_path()).visible_markdown; + self.queue_retained_stream_cell(history_cell::AgentMarkdownCell::new(source, &cwd)); + } + if let Some(source) = retained_sources.1 { + self.queue_retained_stream_cell(history_cell::new_proposed_plan(source, &cwd)); + } + } // 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, diff --git a/codex-rs/tui/src/session_log.rs b/codex-rs/tui/src/session_log.rs index 098bdae2fe..fb005f6e28 100644 --- a/codex-rs/tui/src/session_log.rs +++ b/codex-rs/tui/src/session_log.rs @@ -153,6 +153,15 @@ pub(crate) fn log_inbound_app_event(event: &AppEvent, origin: Option { + let value = json!({ + "ts": now_ts(), + "dir": "to_tui", + "kind": "commit_retained_stream_cell", + "lines": cell.transcript_lines(u16::MAX).len(), + }); + write_inbound_app_event(value, origin); + } AppEvent::StartFileSearch(query) => { let value = json!({ "ts": now_ts(), diff --git a/codex-rs/tui/src/streaming/controller.rs b/codex-rs/tui/src/streaming/controller.rs index d27b5160df..89facda6c3 100644 --- a/codex-rs/tui/src/streaming/controller.rs +++ b/codex-rs/tui/src/streaming/controller.rs @@ -1,8 +1,10 @@ -//! Two-region streaming controllers for agent messages and proposed plans. +//! Streaming controllers for agent messages and proposed plans. //! -//! Each stream partitions rendered markdown into a *stable region* (committed +//! Inline streams partition rendered markdown into a *stable region* (committed //! to scrollback via the animation queue in `StreamState`) and a *tail region* //! (mutable, displayed in the active-cell slot as a transient stream-tail cell). +//! Retained streams expose their complete source as one mutable entry for the +//! application-owned viewport to render and replace in place. //! //! `StreamCore` owns the shared bookkeeping: source accumulation, re-rendering, //! stable/tail partitioning, commit-animation queue management, and terminal @@ -16,16 +18,18 @@ //! (`table_holdback_state`) detects pipe-table patterns (header + delimiter //! pair) in the accumulated source and keeps content from the table header //! onward as mutable tail until the stream finalizes. Holdback is enabled for -//! agent and proposed-plan streams. Lines in `Outside` and `Markdown` fence -//! contexts are scanned; lines inside non-markdown fences are skipped. +//! inline agent and proposed-plan streams. Retained streams keep the full +//! source mutable and do not need table detection. Lines in `Outside` and +//! `Markdown` fence contexts are scanned; lines inside non-markdown fences are +//! skipped. //! //! ## Resize handling //! //! On terminal width change, `StreamCore::set_width` re-renders at the new -//! width and rebuilds the queued stable region from the current emitted line -//! count. This intentionally avoids byte-level remap complexity while the -//! stream is active; finalized content is canonicalized by transcript -//! consolidation into source-backed markdown cells. +//! width. Inline streams rebuild the queued stable region from the current +//! emitted line count; retained streams keep only source and let their live +//! canonical cell re-render. Finalized content is canonicalized into +//! source-backed markdown cells. //! //! ## Invariants //! @@ -59,6 +63,7 @@ use std::time::Duration; use std::time::Instant; use super::StreamState; +use super::StreamSurface; use super::table_holdback::TableHoldbackScanner; use super::table_holdback::TableHoldbackState; #[cfg(test)] @@ -68,21 +73,25 @@ use super::table_holdback::table_holdback_state; // StreamCore — shared bookkeeping for both stream controllers // --------------------------------------------------------------------------- -/// Shared state and logic for the two-region streaming model. +/// Shared state and logic for inline and retained streaming models. /// /// Both [`StreamController`] (agent messages) and [`PlanStreamController`] /// (proposed plans) delegate their core bookkeeping here: source /// accumulation, re-rendering, stable/tail partitioning, commit-animation -/// queue management, and terminal resize handling. +/// queue management, and terminal resize handling. A retained surface keeps +/// source only; its application-owned cell handles live rendering. /// /// The wrapping controllers add only their own `emit()` styling and /// finalize return types. struct StreamCore { state: StreamState, + surface: StreamSurface, /// Current rendering width (columns available for markdown content). width: Option, /// Accumulated raw markdown source for the current stream. raw_source: String, + /// Revision of newline-completed source exposed by a retained stream. + retained_source_revision: u64, /// Full re-render of `raw_source` at `width`. Rebuilt on every committed delta. rendered_lines: Vec, /// Lines enqueued into the commit-animation queue. @@ -120,11 +129,18 @@ struct StreamSelectionProjection { } impl StreamCore { - fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { + fn new( + width: Option, + cwd: &Path, + render_mode: HistoryRenderMode, + surface: StreamSurface, + ) -> Self { Self { state: StreamState::new(width, cwd), + surface, width, raw_source: String::with_capacity(1024), + retained_source_revision: 0, rendered_lines: Vec::with_capacity(64), enqueued_stable_len: 0, emitted_stable_len: 0, @@ -153,7 +169,10 @@ impl StreamCore { if delta.contains('\n') && let Some(committed_source) = self.state.collector.commit_complete_source() { - self.raw_source.push_str(&committed_source); + self.append_completed_source(&committed_source); + if self.surface.is_retained() { + return false; + } self.holdback_scanner.push_source_chunk(&committed_source); self.recompute_streaming_render(); enqueued = self.sync_stable_queue(); @@ -161,18 +180,24 @@ impl StreamCore { enqueued } - /// Drain the collector, render the final source snapshot, and return lines not yet emitted. + /// Drain the collector and return inline lines not yet emitted. /// - /// This intentionally re-renders from the full raw source instead of + /// Inline mode intentionally re-renders from the full raw source instead of /// trying to stitch together queued stable lines and the current tail. The /// final render is the canonical transcript representation used for /// consolidation, so callers that skip `reset()` can accidentally replay a - /// finished stream into the next answer. + /// finished stream into the next answer. Retained mode returns no lines; + /// its source-backed cell performs the canonical render. fn finalize_remaining(&mut self) -> Vec { let remainder_source = self.state.collector.finalize_and_drain_source(); if !remainder_source.is_empty() { - self.raw_source.push_str(&remainder_source); + self.append_completed_source(&remainder_source); + if self.surface.is_retained() { + return Vec::new(); + } self.holdback_scanner.push_source_chunk(&remainder_source); + } else if self.surface.is_retained() { + return Vec::new(); } self.rendered_lines = self.render_source(&self.raw_source); self.selection_projection_cache.take(); @@ -221,13 +246,15 @@ impl StreamCore { self.state.oldest_queued_age(now) } - /// Lines that belong to the mutable tail, not yet queued for stable commit. + /// Inline lines that belong to the mutable tail, not yet queued for stable commit. /// /// The tail starts at `enqueued_stable_len`, so this returns the portion /// of the current render snapshot that is still allowed to change without /// violating scrollback ordering. If callers were to derive the tail from /// `emitted_stable_len` instead, queued-but-not-yet-emitted lines could /// reappear in the active cell and duplicate content on screen. + /// Retained callers render [`Self::retained_live_source`] directly, so no + /// duplicate controller-owned render is maintained for that surface. #[inline] fn current_tail_lines(&self) -> Vec { let start = self.enqueued_stable_len.min(self.rendered_lines.len()); @@ -250,12 +277,31 @@ impl StreamCore { } fn current_tail_selection_fragment(&self) -> Option { + if self.surface.is_retained() { + return None; + } self.selection_fragment(self.enqueued_stable_len..self.rendered_lines.len()) } + fn retained_live_source(&self) -> Option<&str> { + self.surface + .is_retained() + .then_some(self.raw_source.as_str()) + .filter(|source| !source.is_empty()) + } + + fn retained_source_revision(&self) -> Option { + self.retained_live_source() + .map(|_| self.retained_source_revision) + } + #[inline] fn has_tail(&self) -> bool { - self.enqueued_stable_len < self.rendered_lines.len() + if self.surface.is_retained() { + !self.raw_source.is_empty() + } else { + self.enqueued_stable_len < self.rendered_lines.len() + } } /// Update rendering width and rebuild queued stable lines for the new layout. @@ -271,10 +317,13 @@ impl StreamCore { if self.width == width { return; } - let had_pending_queue = self.state.queued_len() > 0; - let had_live_tail = self.has_tail(); self.width = width; self.state.collector.set_width(width); + if self.surface.is_retained() { + return; + } + let had_pending_queue = self.state.queued_len() > 0; + let had_live_tail = self.has_tail(); if self.raw_source.is_empty() { return; } @@ -305,6 +354,7 @@ impl StreamCore { fn reset(&mut self) { self.state.clear(); self.raw_source.clear(); + self.retained_source_revision = 0; self.rendered_lines.clear(); self.enqueued_stable_len = 0; self.emitted_stable_len = 0; @@ -313,6 +363,17 @@ impl StreamCore { self.holdback_scanner.reset(); } + fn append_completed_source(&mut self, source: &str) { + if source.is_empty() { + return; + } + self.raw_source.push_str(source); + if self.surface.is_retained() { + self.retained_source_revision = + self.retained_source_revision.saturating_add(/*rhs*/ 1); + } + } + fn render_source(&self, source: &str) -> Vec { match self.render_mode { HistoryRenderMode::Rich => render_markdown_agent_with_links_and_cwd( @@ -394,9 +455,12 @@ impl StreamCore { return; } + self.render_mode = render_mode; + if self.surface.is_retained() { + return; + } let had_pending_queue = self.state.queued_len() > 0; let had_live_tail = self.has_tail(); - self.render_mode = render_mode; if self.raw_source.is_empty() { return; } @@ -419,6 +483,9 @@ impl StreamCore { /// Compute how many rendered lines should be in the stable region. fn compute_target_stable_len(&mut self) -> usize { + if self.surface.is_retained() { + return 0; + } let tail_budget = self.active_tail_budget_lines(); self.rendered_lines .len() @@ -569,9 +636,14 @@ impl StreamController { /// `width` is the content width available to markdown rendering, not necessarily the full /// terminal width. Passing a stale width after resize will keep queued live output wrapped for /// the old viewport until app-level reflow repairs the finalized transcript. - pub(crate) fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { + pub(crate) fn new( + width: Option, + cwd: &Path, + render_mode: HistoryRenderMode, + surface: StreamSurface, + ) -> Self { Self { - core: StreamCore::new(width, cwd, render_mode), + core: StreamCore::new(width, cwd, render_mode, surface), header_emitted: false, } } @@ -589,18 +661,28 @@ impl StreamController { self.core.reset(); return (None, None); } - let selection_fragment = self - .core - .selection_fragment(rendered_start..self.core.rendered_lines.len()); + let selection_fragment = if self.core.surface.is_retained() { + None + } else { + self.core + .selection_fragment(rendered_start..self.core.rendered_lines.len()) + }; // Move ownership — source is consumed before reset() clears it. let source = std::mem::take(&mut self.core.raw_source); - let out = self.emit(remaining, selection_fragment); + let out = if self.core.surface.is_retained() { + None + } else { + self.emit(remaining, selection_fragment) + }; self.core.reset(); (out, Some(source)) } pub(crate) fn on_commit_tick(&mut self) -> (Option>, bool) { + if self.core.surface.is_retained() { + return (None, true); + } let rendered_start = self.core.emitted_stable_len; let step = self.core.tick(); let selection_fragment = self @@ -613,6 +695,9 @@ impl StreamController { &mut self, max_lines: usize, ) -> (Option>, bool) { + if self.core.surface.is_retained() { + return (None, true); + } let rendered_start = self.core.emitted_stable_len; let step = self.core.tick_batch(max_lines); let selection_fragment = self @@ -643,6 +728,16 @@ impl StreamController { self.core.current_tail_selection_fragment() } + /// Completed source represented by the current retained live display. + pub(crate) fn retained_live_source(&self) -> Option<&str> { + self.core.retained_live_source() + } + + /// Revision of the current completed retained source. + pub(crate) fn retained_source_revision(&self) -> Option { + self.core.retained_source_revision() + } + #[inline] pub(crate) fn tail_starts_stream(&self) -> bool { !self.header_emitted && self.core.enqueued_stable_len == 0 @@ -718,9 +813,14 @@ impl PlanStreamController { /// /// The width has the same meaning as in `StreamController`: it is the markdown body width, and /// callers must update it when the terminal width changes. - pub(crate) fn new(width: Option, cwd: &Path, render_mode: HistoryRenderMode) -> Self { + pub(crate) fn new( + width: Option, + cwd: &Path, + render_mode: HistoryRenderMode, + surface: StreamSurface, + ) -> Self { Self { - core: StreamCore::new(width, cwd, render_mode), + core: StreamCore::new(width, cwd, render_mode, surface), header_emitted: false, top_padding_emitted: false, } @@ -739,22 +839,32 @@ impl PlanStreamController { self.core.reset(); return (None, None); } - let selection_fragment = self - .core - .selection_fragment(rendered_start..self.core.rendered_lines.len()); + let selection_fragment = if self.core.surface.is_retained() { + None + } else { + self.core + .selection_fragment(rendered_start..self.core.rendered_lines.len()) + }; // Move ownership — source is consumed before reset() clears it. let source = std::mem::take(&mut self.core.raw_source); - let out = self.emit( - remaining, - selection_fragment, - /*include_bottom_padding*/ true, - ); + let out = if self.core.surface.is_retained() { + None + } else { + self.emit( + remaining, + selection_fragment, + /*include_bottom_padding*/ true, + ) + }; self.core.reset(); (out, Some(source)) } pub(crate) fn on_commit_tick(&mut self) -> (Option>, bool) { + if self.core.surface.is_retained() { + return (None, true); + } let rendered_start = self.core.emitted_stable_len; let step = self.core.tick(); let selection_fragment = self @@ -774,6 +884,9 @@ impl PlanStreamController { &mut self, max_lines: usize, ) -> (Option>, bool) { + if self.core.surface.is_retained() { + return (None, true); + } let rendered_start = self.core.emitted_stable_len; let step = self.core.tick_batch(max_lines); let selection_fragment = self @@ -822,6 +935,16 @@ impl PlanStreamController { }) } + /// Completed source represented by the current retained live display. + pub(crate) fn retained_live_source(&self) -> Option<&str> { + self.core.retained_live_source() + } + + /// Revision of the current completed retained source. + pub(crate) fn retained_source_revision(&self) -> Option { + self.core.retained_source_revision() + } + pub(crate) fn oldest_queued_age(&self, now: Instant) -> Option { self.core.oldest_queued_age(now) } @@ -915,11 +1038,30 @@ mod tests { } fn stream_controller(width: Option) -> StreamController { - StreamController::new(width, &test_cwd(), HistoryRenderMode::Rich) + StreamController::new( + width, + &test_cwd(), + HistoryRenderMode::Rich, + StreamSurface::Inline, + ) } fn plan_stream_controller(width: Option) -> PlanStreamController { - PlanStreamController::new(width, &test_cwd(), HistoryRenderMode::Rich) + PlanStreamController::new( + width, + &test_cwd(), + HistoryRenderMode::Rich, + StreamSurface::Inline, + ) + } + + fn retained_stream_controller(width: Option) -> StreamController { + StreamController::new( + width, + &test_cwd(), + HistoryRenderMode::Rich, + StreamSurface::Retained, + ) } fn lines_to_plain_strings(lines: &[ratatui::text::Line<'_>]) -> Vec { @@ -984,6 +1126,59 @@ mod tests { lines_to_plain_strings(&lines) } + #[test] + fn retained_agent_stream_keeps_source_without_provisional_cells() { + let width = Some(48); + let mut ctrl = retained_stream_controller(width); + + assert_eq!(ctrl.retained_live_source(), None); + assert_eq!(ctrl.retained_source_revision(), None); + assert!(!ctrl.push("incomplete")); + assert_eq!(ctrl.retained_live_source(), None); + assert_eq!(ctrl.retained_source_revision(), None); + assert_eq!(ctrl.queued_lines(), 0); + assert!(ctrl.current_tail_lines().is_empty()); + assert!(ctrl.current_tail_selection_fragment().is_none()); + + assert!(!ctrl.push(" line\n")); + let first_source = "incomplete line\n"; + assert_eq!(ctrl.retained_live_source(), Some(first_source)); + assert_eq!(ctrl.retained_source_revision(), Some(1)); + + assert!(!ctrl.push("another incomplete")); + assert_eq!(ctrl.retained_live_source(), Some(first_source)); + assert_eq!(ctrl.retained_source_revision(), Some(1)); + + ctrl.set_width(/*width*/ Some(24)); + ctrl.set_render_mode(HistoryRenderMode::Raw); + assert_eq!(ctrl.retained_live_source(), Some(first_source)); + assert_eq!(ctrl.retained_source_revision(), Some(1)); + + let table = " line\n\n| Key | Value |\n| --- | --- |\n| first | short |\n"; + assert!(!ctrl.push(table)); + let source = format!("{first_source}another incomplete{table}"); + assert_eq!(ctrl.retained_live_source(), Some(source.as_str())); + assert_eq!(ctrl.retained_source_revision(), Some(2)); + assert!(ctrl.has_live_tail()); + assert_eq!(ctrl.queued_lines(), 0); + assert!(ctrl.current_tail_lines().is_empty()); + assert!(ctrl.current_tail_selection_fragment().is_none()); + + let (tick_cell, idle) = ctrl.on_commit_tick(); + assert!(tick_cell.is_none()); + assert!(idle); + assert_eq!(ctrl.queued_lines(), 0); + + let (final_cell, finalized_source) = ctrl.finalize(); + assert!(final_cell.is_none()); + assert_eq!(finalized_source.as_deref(), Some(source.as_str())); + assert_eq!(ctrl.retained_live_source(), None); + assert_eq!(ctrl.retained_source_revision(), None); + assert_eq!(ctrl.queued_lines(), 0); + assert!(ctrl.current_tail_lines().is_empty()); + assert!(ctrl.current_tail_selection_fragment().is_none()); + } + #[test] fn controller_set_width_rebuilds_queued_lines() { let mut ctrl = stream_controller(Some(120)); @@ -1935,6 +2130,7 @@ mod tests { /*width*/ Some(32), &test_cwd(), HistoryRenderMode::Rich, + StreamSurface::Inline, ); ctrl.push(&source); diff --git a/codex-rs/tui/src/streaming/mod.rs b/codex-rs/tui/src/streaming/mod.rs index 576277aa3e..f7d9438f6f 100644 --- a/codex-rs/tui/src/streaming/mod.rs +++ b/codex-rs/tui/src/streaming/mod.rs @@ -19,8 +19,11 @@ use crate::terminal_hyperlinks::HyperlinkLine; pub(crate) mod chunking; pub(crate) mod commit_tick; pub(crate) mod controller; +mod surface; mod table_holdback; +pub(crate) use surface::StreamSurface; + struct QueuedLine { line: HyperlinkLine, enqueued_at: Instant, diff --git a/codex-rs/tui/src/streaming/surface.rs b/codex-rs/tui/src/streaming/surface.rs new file mode 100644 index 0000000000..35ad79ca71 --- /dev/null +++ b/codex-rs/tui/src/streaming/surface.rs @@ -0,0 +1,21 @@ +//! Rendering surfaces supported by streaming controllers. +//! +//! Inline terminals can only append stable rows, so their controllers retain +//! the existing queue and table-holdback behavior. Retained viewports own the +//! rendered transcript and can replace a live entry in place, so their entire +//! completed source remains mutable until finalization. + +/// Determines whether a stream emits immutable rows or exposes one mutable render. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum StreamSurface { + /// Emit stable rendered rows into terminal scrollback. + Inline, + /// Keep the full completed source mutable in an application-owned viewport. + Retained, +} + +impl StreamSurface { + pub(super) fn is_retained(self) -> bool { + self == Self::Retained + } +}