Fix live stream reflow at the allocated conversation width (#58927)

## Why

Streaming answers and plans need to wrap to the conversation's allocated width, which can differ from the terminal width. Resizing while reading earlier output should not report new activity unless new output has arrived.

## What changed

Reflow live streams when preparing the owned conversation layout, using its allocated width. Update the observed live revision after presentation-only reflow so it does not trigger `New activity`, while preserving notifications for unseen output.

## Testing

Add regression coverage for streaming answer and plan tables across width changes, including a narrow-layout snapshot, preservation of the reading position, and activity notifications when new output arrives before a resize.

GitOrigin-RevId: a9831c5a1a5ffb7477e7868d4ecca430d6ee50f3

codex-rs/tui/src/app/owned_transcript.rs

@@ -236,6 +236,10 @@ impl App {
     ) -> PreparedConversation {
         self.conversation_frame.inset = conversation_inset;
         let area = self.conversation_area(viewport);
+        if let Some((previous, reflowed)) = self.chat_widget.resize_owned_conversation(area.width) {
+            self.transcript_view
+                .rebind_observed_live_revision(previous, reflowed);
+        }
         self.chat_widget.sync_warnings(&self.transcript_cells);
         let motion = MotionMode::from_animations_enabled(
             self.local_settings.tui.animations && self.local_settings.tui.effects.shimmer,

codex-rs/tui/src/app/resize_reflow.rs

@@ -432,10 +432,7 @@ impl App {
     ) -> Result<()> {
         self.flush_native_history(tui);
         if tui.is_owned_screen() {
-            let width = self.transcript_reflow.note_width(size.width);
-            if width.changed || width.initialized {
-                self.chat_widget.on_terminal_resize(size.width);
-            }
+            self.transcript_reflow.note_width(size.width);
             if size != tui.terminal.last_known_screen_size {
                 self.refresh_status_line();
             }

codex-rs/tui/src/app/snapshots/codex_tui__app__tests__allocated_stream_table.snap

@@ -0,0 +1,28 @@
+---
+source: tui/src/app/tests.rs
+expression: normalize_snapshot_paths(frame.as_str())
+---
+history-33
+
+history-34
+
+history-35
+
+history-36
+
+history-37
+
+history-38
+
+history-39
+
+•  Left                  Right
+  ━━━━━━━━━━━━━━━━━━━━  ━━━━━━━━━━━━━━━━━━━━
+   one two three four    red blue green
+   five six seven        white orange
+
+
+› Ask Codex to do anything
+
+  gpt-test default · /tmp/project
+  ? for shortcuts

codex-rs/tui/src/app/tests.rs

@@ -7038,6 +7038,108 @@ fn plain_line_cell(text: impl Into<String>) -> Arc<dyn HistoryCell> {
     Arc::new(PlainHistoryCell::new(vec![Line::from(text.into())])) as Arc<dyn HistoryCell>
 }
 
+#[tokio::test]
+async fn owned_stream_reflows_at_allocated_width_without_reporting_activity() -> Result<()> {
+    let table = "| Left | Right |\n| --- | --- |\n| one two three four five six seven | red blue green white orange |\n";
+    for kind in ["answer", "plan"] {
+        let mut app = make_test_app().await;
+        let id = ThreadId::new();
+        super::owned_transcript::tests::attach_thread(&mut app, id);
+        app.transcript_cells = (0..40)
+            .map(|i| plain_line_cell(format!("history-{i}")))
+            .collect();
+        if kind == "plan" {
+            app.chat_widget
+                .set_feature_enabled(Feature::CollaborationModes, /*enabled*/ true);
+            app.chat_widget.set_collaboration_mask(
+                crate::collaboration_modes::plan_mask(&app.model_catalog).unwrap(),
+            );
+        }
+        let delta = |text: &str| match kind {
+            "plan" => {
+                ServerNotification::PlanDelta(codex_app_server_protocol::PlanDeltaNotification {
+                    thread_id: id.to_string(),
+                    turn_id: kind.into(),
+                    item_id: kind.into(),
+                    delta: text.into(),
+                })
+            }
+            _ => agent_message_delta_notification(id, kind, kind, text),
+        };
+        let mut tui = crate::tui::test_support::make_test_tui()?;
+        tui.set_owned_screen(/*owned*/ true)?;
+        let mut draw = |app: &mut App, width| -> Result<String> {
+            let screen = tui.terminal.last_known_screen_size;
+            app.handle_draw_pre_render(&mut tui, screen)?;
+            let frames = tui.frame_requester();
+            let mut painted = Default::default();
+            tui.draw(screen.height, |frame| {
+                let area = Rect {
+                    width,
+                    ..frame.area()
+                };
+                let prepared = app.prepare_conversation(frame.area(), Some(area), &frames);
+                painted = app.paint_conversation(frame, &prepared);
+            })?;
+            app.finish_conversation(&mut tui, painted)?;
+            Ok(super::owned_transcript::tests::buffer_text(
+                tui.terminal.previous_buffer(),
+            ))
+        };
+        app.chat_widget
+            .handle_server_notification(delta(table), /*replay_kind*/ None);
+        for width in [80, 44, 80] {
+            let frame = draw(&mut app, width)?;
+            let mut expected = Vec::new();
+            let reserved = if kind == "plan" { 4 } else { 2 };
+            crate::markdown::append_markdown_agent(
+                table,
+                Some(usize::from(width) - reserved),
+                &mut expected,
+            );
+            let separator = expected
+                .iter()
+                .map(ToString::to_string)
+                .find(|line| line.contains('━'))
+                .unwrap();
+            assert!(
+                frame.lines().any(|line| line.trim() == separator.trim()),
+                "{kind} at {width}: {frame}"
+            );
+            if kind == "answer" && width == 44 {
+                insta::assert_snapshot!(
+                    "allocated_stream_table",
+                    normalize_snapshot_paths(frame.as_str())
+                );
+            }
+        }
+        app.transcript_view
+            .jump_to_entry(&app.transcript_cells, /*index*/ 8);
+        for width in [80, 44, 80] {
+            let held = draw(&mut app, width)?;
+            assert!(
+                held.contains("history-8")
+                    && held.contains("Back to bottom")
+                    && !held.contains("New activity"),
+                "{kind}: {held}"
+            );
+        }
+        // Deliver output without painting it. The next resize must preserve that unseen revision.
+        app.chat_widget.handle_server_notification(
+            delta("| new row | arrived |\n"),
+            /*replay_kind*/ None,
+        );
+        let changed = draw(&mut app, 44)?;
+        assert!(
+            changed.contains("history-8") && changed.contains("New activity"),
+            "{kind}: {changed}"
+        );
+        app.transcript_view.jump_to_latest();
+        assert_eq!(draw(&mut app, 80)?.matches("new row").count(), 1, "{kind}");
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn app_server_thread_replacement_clears_previous_transcript_before_replay() -> Result<()> {
     let (mut app, mut events, _op_rx) = make_test_app_with_channels().await;

codex-rs/tui/src/chatwidget.rs

@@ -1747,6 +1747,17 @@ impl ChatWidget {
         }
     }
 
+    /// Reflow at the allocated conversation width and report any presentation-only revision change.
+    pub(crate) fn resize_owned_conversation(&mut self, width: u16) -> Option<(u64, u64)> {
+        if self.last_rendered_width.get() == Some(width) {
+            return None;
+        }
+        let previous = self.transcript.active_cell_revision;
+        self.on_terminal_resize(width);
+        let reflowed = self.transcript.active_cell_revision;
+        (previous != reflowed).then_some((previous, reflowed))
+    }
+
     /// Whether an agent message stream is active (not a plan stream).
     pub(crate) fn has_active_agent_stream(&self) -> bool {
         self.stream_controller.is_some()

codex-rs/tui/src/transcript_view/mutations.rs

@@ -5,6 +5,16 @@ use std::ops::Range;
 use super::*;
 
 impl TranscriptView {
+    /// Rebuild the layout of an observed revision without counting the reflow as new output.
+    pub(crate) fn rebind_observed_live_revision(&mut self, observed: u64, updated: u64) {
+        if let Some((_, key)) = &mut self.live_key
+            && key.revision == observed
+        {
+            key.revision = updated;
+            key.cacheable = false;
+        }
+    }
+
     /// Refresh unseen activity before either footer measurement or transcript painting.
     pub(crate) fn sync_history_tail(&mut self, cells: &[Arc<dyn HistoryCell>]) {
         let tail = cells.last().map(EntryKey::cell);
@@ -148,14 +158,11 @@ impl TranscriptView {
         if self.last_tail == Some(tail_key) {
             self.last_tail = preceding.last().map(EntryKey::cell);
         }
-        let following = self.is_following();
-        if let Some((_, key)) = &mut self.live_key {
-            if !following && key.revision != previous_revision {
+        if let Some((_, key)) = self.live_key {
+            if !self.is_following() && key.revision != previous_revision {
                 self.unseen_activity = true;
             }
-            key.revision = hydrated_revision;
-            // Force fresh lines even if the next draw receives this exact hydrated revision.
-            key.cacheable = false;
+            self.rebind_observed_live_revision(key.revision, hydrated_revision);
         }
         if self.snapshot().is_some() {
             return;