openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-30 19:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tui): retain owned-screen stream source

Browse Source
This commit is contained in:
Felipe Coury
2026-07-05 19:57:26 -03:00
parent ba9d8dd59c
commit 7f75cd211d
18 changed files with 696 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/conversation_panes.rs
+9
View File
@@ -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,
codex-rs/tui/src/app/event_dispatch.rs
+7 -2
View File
@@ -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 } => {
codex-rs/tui/src/app/session_lifecycle.rs
+6
View File
@@ -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,
codex-rs/tui/src/app/tests.rs
+2 -2
View File
@@ -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(
codex-rs/tui/src/app_event.rs
+6
View File
@@ -728,6 +728,12 @@ pub(crate) enum AppEvent {
InsertHistoryCell(Box<dyn HistoryCell>),
/// 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<dyn HistoryCell>),
/// Finish buffering initial resume replay after all replay events have been queued.
EndInitialHistoryReplayBuffer,
codex-rs/tui/src/chatwidget.rs
+13 -1
View File
@@ -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<AddCreditsNudgeCreditType>,
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<StreamController>,
// Stream lifecycle controller for proposed plan output.
plan_stream_controller: Option<PlanStreamController>,
pending_stream_consolidations: usize,
pending_stream_commits: usize,
/// Holds the platform clipboard lease so copied text remains available while supported.
clipboard_lease: Option<crate::clipboard_copy::ClipboardLease>,
copy_last_response_binding: Vec<KeyBinding>,
@@ -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;
codex-rs/tui/src/chatwidget/constructor.rs
+2 -1
View File
@@ -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(),
codex-rs/tui/src/chatwidget/streaming.rs
+127 -26
View File
@@ -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::<history_cell::StreamingAgentTailCell>()
|| cell.as_any().is::<history_cell::StreamingPlanTailCell>()
})
self.transcript
.active_cell
.as_ref()
.is_some_and(|cell| match self.stream_surface {
StreamSurface::Inline => {
cell.as_any().is::<history_cell::StreamingAgentTailCell>()
|| cell.as_any().is::<history_cell::StreamingPlanTailCell>()
}
StreamSurface::Retained => {
cell.as_any().is::<history_cell::AgentMarkdownCell>()
|| cell.as_any().is::<history_cell::ProposedPlanCell>()
}
})
}
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::<history_cell::AgentMarkdownCell>());
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::<history_cell::ProposedPlanCell>());
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)));
}
}
codex-rs/tui/src/chatwidget/tests/slash_commands.rs
+4 -3
View File
@@ -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);
codex-rs/tui/src/chatwidget/tests/snapshots/codex_tui__chatwidget__tests__status_and_layout__retained_live_agent_table_reflows.snap
+19
View File
@@ -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
codex-rs/tui/src/chatwidget/tests/status_and_layout.rs
+178 -3
View File
@@ -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<AppEvent>,
) -> Box<dyn HistoryCell> {
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::<history_cell::AgentMarkdownCell>());
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::<history_cell::ProposedPlanCell>());
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::<Vec<_>>();
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");
codex-rs/tui/src/chatwidget/tokens.rs
+11 -13
View File
@@ -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.
codex-rs/tui/src/chatwidget/transcript.rs
+2
View File
@@ -14,6 +14,8 @@ pub(super) struct TranscriptState {
pub(super) active_cell: Option<Box<dyn HistoryCell>>,
/// 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<u64>,
/// Raw markdown of the most recently completed agent response that
/// survived any local thread rollback.
pub(super) last_agent_markdown: Option<String>,
codex-rs/tui/src/chatwidget/turn_runtime.rs
+42 -9
View File
@@ -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,
codex-rs/tui/src/session_log.rs
+9
View File
@@ -153,6 +153,15 @@ pub(crate) fn log_inbound_app_event(event: &AppEvent, origin: Option<Conversatio
});
write_inbound_app_event(value, origin);
}
AppEvent::CommitRetainedStreamCell(cell) => {
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(),
codex-rs/tui/src/streaming/controller.rs
+235 -39
View File
@@ -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<usize>,
/// 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<HyperlinkLine>,
/// Lines enqueued into the commit-animation queue.
@@ -120,11 +129,18 @@ struct StreamSelectionProjection {
}
impl StreamCore {
fn new(width: Option<usize>, cwd: &Path, render_mode: HistoryRenderMode) -> Self {
fn new(
width: Option<usize>,
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<HyperlinkLine> {
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<HyperlinkLine> {
let start = self.enqueued_stable_len.min(self.rendered_lines.len());
@@ -250,12 +277,31 @@ impl StreamCore {
}
fn current_tail_selection_fragment(&self) -> Option<StreamSelectionFragment> {
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<u64> {
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<HyperlinkLine> {
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<usize>, cwd: &Path, render_mode: HistoryRenderMode) -> Self {
pub(crate) fn new(
width: Option<usize>,
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<Box<dyn HistoryCell>>, 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<Box<dyn HistoryCell>>, 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<u64> {
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<usize>, cwd: &Path, render_mode: HistoryRenderMode) -> Self {
pub(crate) fn new(
width: Option<usize>,
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<Box<dyn HistoryCell>>, 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<Box<dyn HistoryCell>>, 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<u64> {
self.core.retained_source_revision()
}
pub(crate) fn oldest_queued_age(&self, now: Instant) -> Option<Duration> {
self.core.oldest_queued_age(now)
}
@@ -915,11 +1038,30 @@ mod tests {
}
fn stream_controller(width: Option<usize>) -> StreamController {
StreamController::new(width, &test_cwd(), HistoryRenderMode::Rich)
StreamController::new(
width,
&test_cwd(),
HistoryRenderMode::Rich,
StreamSurface::Inline,
)
}
fn plan_stream_controller(width: Option<usize>) -> PlanStreamController {
PlanStreamController::new(width, &test_cwd(), HistoryRenderMode::Rich)
PlanStreamController::new(
width,
&test_cwd(),
HistoryRenderMode::Rich,
StreamSurface::Inline,
)
}
fn retained_stream_controller(width: Option<usize>) -> StreamController {
StreamController::new(
width,
&test_cwd(),
HistoryRenderMode::Rich,
StreamSurface::Retained,
)
}
fn lines_to_plain_strings(lines: &[ratatui::text::Line<'_>]) -> Vec<String> {
@@ -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);
codex-rs/tui/src/streaming/mod.rs
+3
View File
@@ -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,
codex-rs/tui/src/streaming/surface.rs
+21
View File
@@ -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
}
}