diff --git a/crates/core/src/session.rs b/crates/core/src/session.rs
index 1f5a2dbb..a3d8d39c 100644
--- a/crates/core/src/session.rs
+++ b/crates/core/src/session.rs
@@ -551,6 +551,17 @@ impl Timeline {
         timeline
     }
 
+    /// Fold records a client already holds, without cloning their payloads:
+    /// a history page refolds the whole held log, whose tool outputs and
+    /// diffs are the bulk of it.
+    pub fn fold_stored<'a>(records: impl IntoIterator<Item = &'a StoredEvent>) -> Self {
+        let mut timeline = Self::default();
+        for record in records {
+            timeline.apply_stored(record);
+        }
+        timeline
+    }
+
     /// Fold one record, remembering whether it carried a shortened output.
     pub fn apply_stored(&mut self, stored: &StoredEvent) {
         if let Some(bytes) = stored.elided
diff --git a/crates/ui/src/chat/mod.rs b/crates/ui/src/chat/mod.rs
index 08d79f44..6c5d996d 100644
--- a/crates/ui/src/chat/mod.rs
+++ b/crates/ui/src/chat/mod.rs
@@ -51,15 +51,15 @@ use crate::window_state::WindowState;
 use self::components::assistant::MdState;
 use self::components::command_panel::CommandPanelCache;
 use self::model::{
-    ListSync, Segment, TimelineContinuity, TurnIndexCache, TurnListItem, TurnRenderArgs,
+    ListSync, RowRenderArgs, Segment, TimelineContinuity, TimelineRow, TurnIndexCache,
     activity_run_duration_ms, displayed_error_text, divergent_served_model,
-    format_elapsed_deciseconds, latest_message_ids, live_activity_segment, live_edit_counts,
-    live_edit_rows, partition_activity_run, plain_text_as_markdown, segment_entries,
+    format_elapsed_deciseconds, latest_message_ids, live_edit_counts, live_edit_rows,
+    partition_activity_run, plain_text_as_markdown, row_of_entry, rows_of_turn, segment_entries,
     start_hub_projects, timeline_overdraw, user_content, user_visible_text, work_log_capsule_label,
     work_log_counts, work_log_outcome,
 };
 use self::residency::{
-    MarkdownEntry, ResidencyInput, ResidencyScope, decide, tail_turn_window, viewport_turn_window,
+    MarkdownEntry, ResidencyInput, ResidencyScope, decide, tail_row_window, viewport_row_window,
 };
 pub(crate) use crate::material::{
     CHAT_CONTENT_MAX_WIDTH as CONTENT_MAX_WIDTH, CHAT_CONTENT_MIN_PADDING as CONTENT_MIN_PADDING,
@@ -71,6 +71,8 @@ const TRAFFIC_LIGHT_INSET: f32 = 80.;
 /// Vertical rhythm between turns. Turns are separated by space and typographic
 /// hierarchy alone — there is deliberately no rule/divider under the user bubble.
 const TURN_GAP: f32 = 32.;
+/// Vertical rhythm between the segments of one turn.
+const SEGMENT_GAP: f32 = 10.;
 /// Large documents are parsed away from the UI executor before becoming resident.
 const ASYNC_MARKDOWN_THRESHOLD_BYTES: usize = 4 * 1024;
 /// Target minimum time for the latest activity and its immediate predecessor.
@@ -330,7 +332,6 @@ struct PendingMarkdownBuild {
     generation: u64,
     session_key: Option<String>,
     desired_text: String,
-    turn: usize,
 }
 
 /// A whole tool output fetched on request.
@@ -362,12 +363,12 @@ pub struct ChatView {
     /// scroll applied between two frames is either overwritten by the next
     /// packet or would have to give up when one arrived first.
     reservation_scroll_back: Option<ReservationScrollBack>,
-    turn_items: Vec<TurnListItem>,
+    rows: Vec<TimelineRow>,
     turn_index_cache: TurnIndexCache,
     md_states: HashMap<String, MdState>,
     pending_md_builds: HashMap<String, PendingMarkdownBuild>,
     next_md_build_generation: u64,
-    markdown_visible_turns: Range<usize>,
+    markdown_visible_rows: Range<usize>,
     markdown_scroll_top: Option<usize>,
     /// Open/closed keys for collapsibles other than activity details.
     expanded: HashSet<String>,
@@ -394,7 +395,7 @@ pub struct ChatView {
     capture: Option<CaptureSession>,
     _subscriptions: Vec<Subscription>,
     #[cfg(test)]
-    markdown_remeasured_turns: Vec<usize>,
+    markdown_remeasured_rows: Vec<usize>,
 }
 
 // ListState retains measured row heights, so variable-height turns use the same
@@ -454,7 +455,7 @@ impl ChatView {
     /// The timeline's on-screen rectangle, when it has something to capture.
     pub(crate) fn capture_viewport(&self) -> Option<gpui::Bounds<gpui::Pixels>> {
         let bounds = self.list_state.viewport_bounds();
-        (!self.turn_items.is_empty() && bounds.size.height > px(0.) && bounds.size.width > px(0.))
+        (!self.rows.is_empty() && bounds.size.height > px(0.) && bounds.size.width > px(0.))
             .then_some(bounds)
     }
 
@@ -583,11 +584,11 @@ impl ChatView {
         list_state.set_follow_mode(FollowMode::Tail);
         let chat = cx.entity().downgrade();
         list_state.set_scroll_handler(move |event, window, cx| {
-            let visible_turns = event.visible_range.clone();
+            let visible_rows = event.visible_range.clone();
             let chat = chat.clone();
             window.defer(cx, move |_, cx| {
                 let _ = chat.update(cx, |chat, cx| {
-                    chat.set_markdown_visible_turns(visible_turns, cx);
+                    chat.set_markdown_visible_rows(visible_rows, cx);
                 });
             });
         });
@@ -626,12 +627,12 @@ impl ChatView {
             timeline_stale: false,
             history_placeholder_height: px(0.),
             reservation_scroll_back: None,
-            turn_items: Vec::new(),
+            rows: Vec::new(),
             turn_index_cache: TurnIndexCache::default(),
             md_states: HashMap::new(),
             pending_md_builds: HashMap::new(),
             next_md_build_generation: 0,
-            markdown_visible_turns: 0..0,
+            markdown_visible_rows: 0..0,
             markdown_scroll_top: None,
             expanded: HashSet::new(),
             auto_activity_expansions: AutoActivityExpansions::default(),
@@ -647,7 +648,7 @@ impl ChatView {
             capture: None,
             _subscriptions: subscriptions,
             #[cfg(test)]
-            markdown_remeasured_turns: Vec::new(),
+            markdown_remeasured_rows: Vec::new(),
         };
         this.sync_markdown_states(cx);
         this
@@ -680,7 +681,7 @@ impl ChatView {
             .read(cx)
             .with_active_timeline(|timeline| {
                 let list_sync = self.turn_index_cache.sync(
-                    &mut self.turn_items,
+                    &mut self.rows,
                     &timeline.turns,
                     &timeline.entries,
                     timeline
@@ -696,7 +697,7 @@ impl ChatView {
             })
             .unwrap_or_else(|| {
                 let list_sync = self.turn_index_cache.sync(
-                    &mut self.turn_items,
+                    &mut self.rows,
                     &[],
                     &[],
                     None,
@@ -723,8 +724,8 @@ impl ChatView {
             self.full_outputs.clear();
             self.highlighted_turn = None;
             self.session_key = session_key;
-            self.markdown_visible_turns = tail_turn_window(self.turn_items.len());
-            self.markdown_scroll_top = Some(self.turn_items.len());
+            self.markdown_visible_rows = tail_row_window(self.rows.len());
+            self.markdown_scroll_top = Some(self.rows.len());
         }
 
         match list_sync {
@@ -779,16 +780,16 @@ impl ChatView {
                     self.list_state.set_follow_mode(FollowMode::Tail);
                 }
             }
-            ListSync::Incremental { append, remeasure } => {
-                if let Some(range) = append {
-                    let count = range.len();
-                    // The former last turn hands its edge padding to the new
-                    // one; its cached height must not keep it.
+            ListSync::Incremental { splices, remeasure } => {
+                for (range, count) in splices.into_iter().rev() {
+                    // The row above a splice changes its padding (the former
+                    // last row hands its edge padding to the new one); its
+                    // cached height must not keep it.
                     if range.start > 0 {
                         self.list_state
                             .remeasure_items(range.start - 1..range.start);
                     }
-                    self.list_state.splice(range.start..range.start, count);
+                    self.list_state.splice(range, count);
                 }
                 for index in remeasure {
                     self.list_state.remeasure_items(index..index + 1);
@@ -797,20 +798,21 @@ impl ChatView {
         }
 
         if self.list_state.is_following_tail() {
-            self.markdown_visible_turns = tail_turn_window(self.turn_items.len());
-            self.markdown_scroll_top = Some(self.turn_items.len());
+            self.markdown_visible_rows = tail_row_window(self.rows.len());
+            self.markdown_scroll_top = Some(self.rows.len());
         }
 
-        if let Some(turn) = requested_turn.filter(|turn| *turn < self.turn_items.len()) {
+        let requested_row = requested_turn.and_then(|turn| rows_of_turn(&self.rows, turn).next());
+        if let (Some(turn), Some(row)) = (requested_turn, requested_row) {
             self.reservation_scroll_back = None;
             self.list_state.pause_following_tail();
             self.list_state.scroll_to(ListOffset {
-                item_ix: turn,
+                item_ix: row,
                 offset_in_item: px(0.),
             });
             self.highlighted_turn = Some(turn);
-            self.markdown_visible_turns = viewport_turn_window(turn, self.turn_items.len());
-            self.markdown_scroll_top = Some(turn);
+            self.markdown_visible_rows = viewport_row_window(row, self.rows.len());
+            self.markdown_scroll_top = Some(row);
             if let Some(session_id) = self.session_key.as_deref() {
                 self.workspace_store.update(cx, |store, _cx| {
                     store.take_pending_chat_turn(session_id, turn);
@@ -818,7 +820,7 @@ impl ChatView {
             }
         }
 
-        self.sync_markdown_residency(requested_turn, cx);
+        self.sync_markdown_residency(requested_row, cx);
 
         // Keep a 100ms ticker alive while a turn runs so the live elapsed timer
         // advances at decisecond precision; dropping it cancels the task.
@@ -876,10 +878,10 @@ impl ChatView {
 
     fn sync_markdown_residency(
         &mut self,
-        one_shot_turn_target: Option<usize>,
+        one_shot_row_target: Option<usize>,
         cx: &mut Context<Self>,
     ) {
-        let turn_count = self.turn_items.len();
+        let row_count = self.rows.len();
         // Auto-scroll can move many rows during a drag. Do not retire any
         // participant until mouse-up; completed-selection participants remain
         // pinned individually below so copy keeps its full projection.
@@ -902,16 +904,16 @@ impl ChatView {
             .read(cx)
             .with_active_timeline(|timeline| {
                 let scope = ResidencyScope::new(
-                    turn_count,
-                    self.markdown_visible_turns.clone(),
-                    one_shot_turn_target,
+                    row_count,
+                    self.markdown_visible_rows.clone(),
+                    one_shot_row_target,
                     timeline.turn_running,
                 );
-                let entries = markdown_entries_for_residency(timeline, &scope).entries;
+                let entries = markdown_entries_for_residency(timeline, &self.rows, &scope).entries;
                 let decisions = decide(ResidencyInput {
-                    turn_count,
-                    visible_turns: self.markdown_visible_turns.clone(),
-                    one_shot_turn_target,
+                    row_count,
+                    visible_rows: self.markdown_visible_rows.clone(),
+                    one_shot_row_target,
                     entries: &entries,
                     stream_running: timeline.turn_running,
                     resident_ids: &resident_ids,
@@ -919,21 +921,22 @@ impl ChatView {
                     selection_drag_active,
                 });
                 let mut texts = Vec::new();
-                for entry in &timeline.entries {
+                for (index, entry) in timeline.entries.iter().enumerate() {
                     if !decisions.build.contains(&entry.id) {
                         continue;
                     }
+                    let row = row_of_entry(&self.rows, index, entry.turn);
                     match &entry.content {
                         EntryContent::Item(ItemContent::AssistantMessage { text })
                         | EntryContent::Item(ItemContent::Reasoning { text }) => {
-                            texts.push((entry.turn, entry.id.clone(), text.clone()));
+                            texts.push((row, entry.id.clone(), text.clone()));
                         }
                         content => {
                             let Some((text, _, context_len, _)) = user_content(content) else {
                                 continue;
                             };
                             texts.push((
-                                entry.turn,
+                                row,
                                 entry.id.clone(),
                                 plain_text_as_markdown(user_visible_text(text, context_len)),
                             ));
@@ -943,7 +946,11 @@ impl ChatView {
                 if let Some(plan) = &timeline.proposed_plan {
                     let id = format!("plan:{}", plan.item_id);
                     if decisions.build.contains(&id) {
-                        texts.push((plan.turn, id, plan.markdown.clone()));
+                        texts.push((
+                            rows_of_turn(&self.rows, plan.turn).last(),
+                            id,
+                            plan.markdown.clone(),
+                        ));
                     }
                 }
                 (texts, decisions)
@@ -953,8 +960,8 @@ impl ChatView {
         self.pending_md_builds
             .retain(|id, _| decisions.build.contains(id) && !decisions.evict.contains(id));
 
-        let mut rebuilt_turns = HashSet::new();
-        for (turn, id, text) in texts {
+        let mut rebuilt_rows = HashSet::new();
+        for (row, id, text) in texts {
             match self.md_states.get_mut(&id) {
                 Some(md) => md.sync(text, cx),
                 None if self.pending_md_builds.contains_key(&id) => {
@@ -963,47 +970,63 @@ impl ChatView {
                         .get_mut(&id)
                         .expect("pending Markdown build disappeared");
                     pending.desired_text = text;
-                    pending.turn = turn;
                 }
                 None if text.len() > ASYNC_MARKDOWN_THRESHOLD_BYTES => {
-                    self.spawn_markdown_build(turn, id, text, cx);
+                    self.spawn_markdown_build(id, text, cx);
                 }
                 None => {
                     self.md_states.insert(id, MdState::new(&text, cx));
-                    rebuilt_turns.insert(turn);
+                    rebuilt_rows.extend(row);
                 }
             }
         }
-        let mut rebuilt_turns = rebuilt_turns
+        let mut rebuilt_rows = rebuilt_rows
             .into_iter()
-            .filter(|turn| *turn < self.turn_items.len())
+            .filter(|row| *row < self.rows.len())
             .collect::<Vec<_>>();
-        rebuilt_turns.sort_unstable();
-        let Some((&first, rest)) = rebuilt_turns.split_first() else {
+        rebuilt_rows.sort_unstable();
+        let Some((&first, rest)) = rebuilt_rows.split_first() else {
             return;
         };
         let mut range = first..first + 1;
-        for &turn in rest {
-            if turn == range.end {
+        for &row in rest {
+            if row == range.end {
                 range.end += 1;
             } else {
                 // Eviction leaves this cache untouched. A lazy rebuild only
                 // invalidates rebuilt rows; ListState preserves the absolute
                 // scroll-top offset while it measures the parsed Markdown.
-                self.remeasure_markdown_turns(range);
-                range = turn..turn + 1;
+                self.remeasure_markdown_rows(range);
+                range = row..row + 1;
             }
         }
-        self.remeasure_markdown_turns(range);
+        self.remeasure_markdown_rows(range);
     }
 
-    fn spawn_markdown_build(
-        &mut self,
-        turn: usize,
-        id: String,
-        text: String,
-        cx: &mut Context<Self>,
-    ) {
+    /// The row rendering the Markdown document `id` (an entry id, or
+    /// `plan:<item>` for the proposed plan), looked up when it is needed:
+    /// rows shift while a build runs, so a build carries no row.
+    fn markdown_row(&self, id: &str, cx: &App) -> Option<usize> {
+        self.workspace_store
+            .read(cx)
+            .with_active_timeline(|timeline| {
+                if let Some(item_id) = id.strip_prefix("plan:") {
+                    return timeline
+                        .proposed_plan
+                        .as_ref()
+                        .filter(|plan| plan.item_id == item_id)
+                        .and_then(|plan| rows_of_turn(&self.rows, plan.turn).last());
+                }
+                timeline
+                    .entries
+                    .iter()
+                    .position(|entry| entry.id == id)
+                    .and_then(|index| row_of_entry(&self.rows, index, timeline.entries[index].turn))
+            })
+            .flatten()
+    }
+
+    fn spawn_markdown_build(&mut self, id: String, text: String, cx: &mut Context<Self>) {
         self.next_md_build_generation = self.next_md_build_generation.wrapping_add(1);
         let generation = self.next_md_build_generation;
         self.pending_md_builds.insert(
@@ -1012,7 +1035,6 @@ impl ChatView {
                 generation,
                 session_key: self.session_key.clone(),
                 desired_text: text.clone(),
-                turn,
             },
         );
         let parse_text = text;
@@ -1044,7 +1066,6 @@ impl ChatView {
             return;
         }
         let desired_text = pending.desired_text.clone();
-        let turn = pending.turn;
         self.pending_md_builds.remove(&id);
 
         if desired_text != parsed_text && !desired_text.starts_with(&parsed_text) {
@@ -1058,42 +1079,51 @@ impl ChatView {
         if desired_text != parsed_text {
             state.sync(desired_text, cx);
         }
+        let row = self.markdown_row(&id, cx);
         self.md_states.insert(id, state);
-        if turn < self.turn_items.len() {
-            self.remeasure_markdown_turns(turn..turn + 1);
+        if let Some(row) = row.filter(|row| *row < self.rows.len()) {
+            self.remeasure_markdown_rows(row..row + 1);
         }
         cx.notify();
     }
 
-    fn remeasure_markdown_turns(&mut self, range: Range<usize>) {
+    fn remeasure_markdown_rows(&mut self, range: Range<usize>) {
         #[cfg(test)]
-        self.markdown_remeasured_turns.extend(range.clone());
+        self.markdown_remeasured_rows.extend(range.clone());
         self.list_state.remeasure_items(range);
     }
 
-    fn set_markdown_visible_turns(&mut self, visible_turns: Range<usize>, cx: &mut Context<Self>) {
-        let turn_count = self.turn_items.len();
-        let visible_turns = visible_turns.start.min(turn_count)..visible_turns.end.min(turn_count);
-        self.markdown_scroll_top = Some(visible_turns.start);
-        if visible_turns == self.markdown_visible_turns {
+    /// Invalidate the measured heights of every row of `turn`.
+    fn remeasure_turn(&self, turn: usize) {
+        let rows = rows_of_turn(&self.rows, turn);
+        if !rows.is_empty() {
+            self.list_state.remeasure_items(rows);
+        }
+    }
+
+    fn set_markdown_visible_rows(&mut self, visible_rows: Range<usize>, cx: &mut Context<Self>) {
+        let row_count = self.rows.len();
+        let visible_rows = visible_rows.start.min(row_count)..visible_rows.end.min(row_count);
+        self.markdown_scroll_top = Some(visible_rows.start);
+        if visible_rows == self.markdown_visible_rows {
             return;
         }
-        self.markdown_visible_turns = visible_turns;
+        self.markdown_visible_rows = visible_rows;
         self.sync_markdown_residency(None, cx);
         cx.notify();
     }
 
     fn sync_markdown_scroll_position(&mut self, cx: &mut Context<Self>) {
-        let turn_count = self.turn_items.len();
-        let scroll_top = self.list_state.logical_scroll_top().item_ix.min(turn_count);
+        let row_count = self.rows.len();
+        let scroll_top = self.list_state.logical_scroll_top().item_ix.min(row_count);
         if self.markdown_scroll_top == Some(scroll_top) {
             return;
         }
         self.markdown_scroll_top = Some(scroll_top);
-        self.markdown_visible_turns = if scroll_top == turn_count {
-            tail_turn_window(turn_count)
+        self.markdown_visible_rows = if scroll_top == row_count {
+            tail_row_window(row_count)
         } else {
-            viewport_turn_window(scroll_top, turn_count)
+            viewport_row_window(scroll_top, row_count)
         };
         self.sync_markdown_residency(None, cx);
     }
@@ -1144,7 +1174,7 @@ impl ChatView {
         // remeasure covers collapsibles whose state is intentionally not
         // fingerprinted.
         self.timeline_stale = true;
-        self.list_state.remeasure_items(turn..turn + 1);
+        self.remeasure_turn(turn);
         cx.notify();
     }
 
@@ -1179,7 +1209,7 @@ impl ChatView {
                         generation,
                     ) && this.session_key.as_deref() == Some(collapse_session_key.as_str())
                     {
-                        this.list_state.remeasure_items(turn..turn + 1);
+                        this.remeasure_turn(turn);
                         cx.notify();
                     }
                 });
@@ -1189,30 +1219,30 @@ impl ChatView {
         auto.expanded
     }
 
-    /// Render one turn as chronological messages, errors, and Work Log runs.
+    /// Render one timeline row: the segment it spans (a message, an error, a
+    /// Work Log run) and, on the turn's last row, the turn's trailer.
     ///
-    /// `pinned` carries the ids of the last user / last assistant message in the
-    /// whole timeline: their action rows stay visible instead of waiting for a
-    /// hover, so Copy is never invisible-and-hover-only.
-    fn render_turn(
+    /// `entries` are the turn's entries; the row renders its own range of
+    /// them and the trailer reads the rest (pending steers, the last
+    /// timestamp). `pinned` carries the ids of the last user / last assistant
+    /// message in the whole timeline: their action rows stay visible instead
+    /// of waiting for a hover, so Copy is never invisible-and-hover-only.
+    fn render_row(
         &mut self,
-        args: TurnRenderArgs<'_>,
+        args: RowRenderArgs<'_>,
         window: &mut Window,
         cx: &mut Context<Self>,
     ) -> AnyElement {
-        let (index, turn, cwd, entries, pinned) = args;
-        let mut column = v_flex().w_full().gap(px(10.));
+        let (row, turn, cwd, entries, pinned) = args;
+        let index = row.turn;
+        let mut column = v_flex().w_full().gap(px(SEGMENT_GAP));
 
-        let segmented = segment_entries(entries, turn.running);
+        // A row's range re-segments to the one segment it was indexed from.
+        let own = entries.get(row.entry_range.clone()).unwrap_or(&[]);
+        let segmented = segment_entries(own, turn.running);
         let segments = &segmented.flow;
-        let last_activity_segment = live_activity_segment(segments, turn.running);
-        // Only the turn's final assistant text is the deliverable; interim
-        // notes between tool runs carry no action row.
-        let last_assistant_segment = segments
-            .iter()
-            .rposition(|segment| matches!(segment, Segment::Assistant(_)));
 
-        for (segment_index, segment) in segments.iter().enumerate() {
+        for segment in segments.iter() {
             match segment {
                 Segment::Relay(entry) => {
                     let EntryContent::ProviderRelay {
@@ -1263,14 +1293,7 @@ impl ChatView {
                 Segment::ActivityRun(activities) => {
                     let segment_id = activities[0].id.as_str();
                     column = column.child(self.compose_work_log(
-                        (
-                            index,
-                            segment_id,
-                            turn,
-                            cwd,
-                            activities,
-                            last_activity_segment == Some(segment_index),
-                        ),
+                        (index, segment_id, turn, cwd, activities, row.live_activity),
                         cx,
                     ));
                 }
@@ -1324,8 +1347,7 @@ impl ChatView {
                             markdown,
                             compact: self.window_state.read(cx).compact,
                             pinned: pinned.1 == Some(entry.id.as_str()),
-                            show_actions: !turn.running
-                                && last_assistant_segment == Some(segment_index),
+                            show_actions: !turn.running && row.last_assistant,
                             copied,
                         },
                         cx.listener(move |this, _, _, cx| {
@@ -1397,6 +1419,10 @@ impl ChatView {
             }
         }
 
+        if !row.last_in_turn {
+            return column.into_any_element();
+        }
+
         if let Some((item_id, markdown)) = self
             .workspace_store
             .read(cx)
@@ -1481,9 +1507,10 @@ impl ChatView {
         }
 
         // Pending steers float below every live transcript/work-log element.
-        // Keeping them separate from `segments` preserves FIFO order without
-        // making their request-time position look model-visible.
-        for entry in segmented.pending_steers {
+        // They are read from the whole turn, not this row's segment, so FIFO
+        // order holds without making their request-time position look
+        // model-visible.
+        for entry in segment_entries(entries, turn.running).pending_steers {
             let EntryContent::Steer {
                 text,
                 status,
@@ -1958,7 +1985,7 @@ impl ChatView {
                     .borrow_mut()
                     .adopt(&id, generation, frame)
                 {
-                    this.list_state.remeasure_items(turn..turn + 1);
+                    this.remeasure_turn(turn);
                 }
                 cx.notify();
             });
@@ -2778,15 +2805,15 @@ struct ResidencyMarkdownEntries {
 
 fn markdown_entries_for_residency(
     timeline: &Timeline,
+    rows: &[TimelineRow],
     scope: &ResidencyScope,
 ) -> ResidencyMarkdownEntries {
     let mut entries = Vec::new();
-    for entry in &timeline.entries {
-        let turn_running = timeline
-            .turns
-            .get(entry.turn)
-            .is_some_and(|turn| turn.running);
-        if !scope.includes(entry.turn, turn_running) {
+    for (index, entry) in timeline.entries.iter().enumerate() {
+        let Some(row) = row_of_entry(rows, index, entry.turn) else {
+            continue;
+        };
+        if !scope.includes(row) {
             continue;
         }
         let markdown_bearing = matches!(
@@ -2797,23 +2824,18 @@ fn markdown_entries_for_residency(
         if markdown_bearing {
             entries.push(MarkdownEntry {
                 id: entry.id.clone(),
-                turn: entry.turn,
-                turn_running,
+                row,
             });
         }
     }
-    if let Some(plan) = &timeline.proposed_plan {
-        let turn_running = timeline
-            .turns
-            .get(plan.turn)
-            .is_some_and(|turn| turn.running);
-        if scope.includes(plan.turn, turn_running) {
-            entries.push(MarkdownEntry {
-                id: format!("plan:{}", plan.item_id),
-                turn: plan.turn,
-                turn_running,
-            });
-        }
+    if let Some(plan) = &timeline.proposed_plan
+        && let Some(row) = rows_of_turn(rows, plan.turn).last()
+        && scope.includes(row)
+    {
+        entries.push(MarkdownEntry {
+            id: format!("plan:{}", plan.item_id),
+            row,
+        });
     }
     ResidencyMarkdownEntries {
         #[cfg(test)]
@@ -2897,7 +2919,7 @@ impl Render for ChatView {
         } else {
             px(0.)
         };
-        if placeholder != self.history_placeholder_height && !self.turn_items.is_empty() {
+        if placeholder != self.history_placeholder_height && !self.rows.is_empty() {
             let anchor = self.list_state.logical_scroll_top();
             let following = self.list_state.is_following_tail();
             self.list_state.remeasure_items(0..1);
@@ -2968,30 +2990,37 @@ impl Render for ChatView {
             .with_active_timeline(|timeline| latest_message_ids(&timeline.entries))
             .unwrap_or_default();
 
-        let item_count = self.turn_items.len();
+        let item_count = self.rows.len();
         let item_cwd = cwd.clone();
         let timeline = list(
             self.list_state.clone(),
             cx.processor(move |this, index: usize, window, cx| {
-                let Some(item) = this.turn_items.get(index) else {
+                let Some(row) = this.rows.get(index).cloned() else {
                     return div().into_any_element();
                 };
-                // Clone only the handful of entries in this visible/overdrawn
-                // turn. The full history remains behind the store and is never
-                // cloned by the render path.
+                // Clone only the entries of this row's turn: the row renders
+                // its own segment and the turn's last row its trailer. The
+                // full history remains behind the store and is never cloned
+                // by the render path.
+                let turn_entries = {
+                    let rows = rows_of_turn(&this.rows, row.turn);
+                    let start = this.rows[rows.start].entry_range.start;
+                    let end = this.rows[rows.end - 1].entry_range.end;
+                    start.min(end)..end
+                };
                 let Some((turn, entries)) =
                     this.workspace_store
                         .read(cx)
                         .with_active_timeline(|timeline| {
                             (
-                                timeline.turns.get(index).cloned().unwrap_or_default(),
-                                // `entry_range` comes from `turn_items`, a snapshot
-                                // that can trail the live timeline by a frame (e.g.
-                                // adopting a running background thread whose timeline
-                                // is being re-folded), so it must not index blindly.
+                                timeline.turns.get(row.turn).cloned().unwrap_or_default(),
+                                // The rows are a snapshot that can trail the live
+                                // timeline by a frame (e.g. adopting a running
+                                // background thread whose timeline is being
+                                // re-folded), so they must not index blindly.
                                 timeline
                                     .entries
-                                    .get(item.entry_range.clone())
+                                    .get(turn_entries.clone())
                                     .map(<[_]>::to_vec)
                                     .unwrap_or_default(),
                             )
@@ -2999,9 +3028,15 @@ impl Render for ChatView {
                 else {
                     return div().into_any_element();
                 };
-                let rendered = this.render_turn(
+                // The row's range, relative to the turn's entries.
+                let row = TimelineRow {
+                    entry_range: row.entry_range.start - turn_entries.start
+                        ..row.entry_range.end - turn_entries.start,
+                    ..row
+                };
+                let rendered = this.render_row(
                     (
-                        index,
+                        &row,
                         &turn,
                         &item_cwd,
                         &entries,
@@ -3019,15 +3054,17 @@ impl Render for ChatView {
                     } else {
                         CONTENT_MIN_PADDING
                     }))
-                    .when(this.highlighted_turn == Some(index), |item| {
+                    .when(this.highlighted_turn == Some(row.turn), |item| {
                         item.rounded(crate::material::radius_card())
                             .bg(cx.theme().list_active)
                     })
                     .when(index == 0, |item| item.pt(px(TIMELINE_EDGE_PADDING)))
-                    .pb(px(if index + 1 < item_count {
+                    .pb(px(if index + 1 == item_count {
+                        TIMELINE_EDGE_PADDING
+                    } else if row.last_in_turn {
                         TURN_GAP
                     } else {
-                        TIMELINE_EDGE_PADDING
+                        SEGMENT_GAP
                     }))
                     // `min_w_0`: a turn holds nowrap content (diff rows, command
                     // output). Without it this flex item grows to that content
@@ -3905,14 +3942,34 @@ mod tests {
         let mut timeline = synthetic_markdown_timeline(200);
         timeline.turns[5].running = true;
         timeline.turn_running = true;
-        let scope = ResidencyScope::new(200, 40..48, None, true);
+        // Three rows per synthetic turn: the user bubble, the reasoning
+        // run and the assistant message.
+        let rows = super::model::index_rows(
+            &timeline.turns,
+            &timeline.entries,
+            None,
+            &std::collections::HashSet::new(),
+        );
+        assert_eq!(rows.len(), 600);
+        let scope = ResidencyScope::new(600, 120..144, None, true);
 
-        let candidates = markdown_entries_for_residency(&timeline, &scope);
+        let candidates = markdown_entries_for_residency(&timeline, &rows, &scope);
 
-        assert_eq!(candidates.constructions, 177);
+        assert_eq!(candidates.constructions, 172);
         assert!(candidates.constructions < timeline.entries.len());
-        assert!(candidates.entries.iter().any(|entry| entry.turn == 5));
-        assert!(candidates.entries.iter().any(|entry| entry.turn == 199));
+        // A running turn far from the viewport is history like any other.
+        assert!(
+            !candidates
+                .entries
+                .iter()
+                .any(|entry| entry.id == "assistant-5")
+        );
+        assert!(
+            candidates
+                .entries
+                .iter()
+                .any(|entry| entry.id == "assistant-199")
+        );
     }
 
     #[gpui::test]
@@ -4291,7 +4348,7 @@ mod tests {
                     let _ = window.draw(cx);
                 });
                 let last = cx
-                    .debug_bounds("timeline-row-29")
+                    .debug_bounds("timeline-row-89")
                     .expect("last timeline row");
                 assert!(
                     f32::from(last.bottom() - list.viewport_bounds().bottom()).abs() <= 1.,
@@ -4482,17 +4539,19 @@ mod tests {
         });
         draw(cx);
 
-        let after = list.logical_scroll_top();
+        // The row the reader was on sits 60 rows lower and exactly 40px
+        // further down the viewport: both the page and the pan applied.
+        let row_top = list
+            .bounds_for_item(before.item_ix + 60)
+            .expect("the reader's row stays on screen")
+            .top();
         assert_eq!(
-            after.item_ix,
-            before.item_ix + 20,
-            "a 40px pan moved the reader off turn {} to turn {} (offset {:?} -> {:?})",
-            before.item_ix + 20,
-            after.item_ix,
+            row_top - list.viewport_bounds().top(),
+            px(40.) - before.offset_in_item,
+            "a 40px pan and a 60-row page both apply (offset {:?} -> {:?})",
             before.offset_in_item,
-            after.offset_in_item
+            list.logical_scroll_top().offset_in_item
         );
-        assert_eq!(after.offset_in_item, before.offset_in_item - px(40.));
         assert_eq!(list.scroll_px_offset_for_scrollbar().y, before_px + px(40.));
     }
 
@@ -4637,7 +4696,7 @@ mod tests {
             cx.debug_bounds("scroll-to-end").is_some(),
             "prepend hides jump"
         );
-        assert_eq!(list.logical_scroll_top().item_ix, before.item_ix + 20);
+        assert_eq!(list.logical_scroll_top().item_ix, before.item_ix + 60);
         assert_eq!(
             list.logical_scroll_top().offset_in_item,
             before.offset_in_item
@@ -4798,8 +4857,8 @@ mod tests {
         });
         let anchor = list.logical_scroll_top();
         assert_eq!(
-            anchor.item_ix, 30,
-            "the same turn must remain visible after 20 earlier turns"
+            anchor.item_ix, 70,
+            "the same row must remain visible after 20 earlier turns of three rows"
         );
         assert_eq!(anchor.offset_in_item, px(7.));
         assert!(!list.is_following_tail());
@@ -4864,8 +4923,8 @@ mod tests {
         );
         assert_eq!(settle(px(-400.), cx), px(-400.));
         assert!(
-            list.logical_scroll_top().item_ix < 56,
-            "a screen above the last turns"
+            list.logical_scroll_top().item_ix < 176,
+            "a screen above the last rows"
         );
         assert_eq!(
             settle(px(400.), cx),
@@ -4957,9 +5016,11 @@ mod tests {
             let _ = window.draw(cx);
         });
         let anchor = list.logical_scroll_top();
+        // The page adds the earlier turn's three rows and the user bubble
+        // above the answer.
         assert_eq!(
             (anchor.item_ix, anchor.offset_in_item),
-            (1, px(20.)),
+            (5, px(20.)),
             "the completed turn keeps its reader at the same content offset"
         );
         assert!(
@@ -5027,11 +5088,11 @@ mod tests {
         }
         let anchor = list.logical_scroll_top();
         assert!(
-            anchor.item_ix < 20 && anchor.offset_in_item >= px(0.),
+            anchor.item_ix < 60 && anchor.offset_in_item >= px(0.),
             "the viewport top sits inside the measured page, not above its row: {anchor:?}"
         );
         let after = list
-            .bounds_for_item(20)
+            .bounds_for_item(60)
             .expect("previous first turn remains on screen")
             .top();
         assert!(
@@ -5096,7 +5157,7 @@ mod tests {
             let _ = window.draw(cx);
         });
         let after = list
-            .bounds_for_item(20)
+            .bounds_for_item(60)
             .expect("previous first turn remains on screen")
             .top();
         assert!(
@@ -5169,7 +5230,8 @@ mod tests {
     fn chat_view_applies_markdown_residency_decisions(cx: &mut TestAppContext) {
         use gpui::{FollowMode, ListOffset, VisualTestContext, px, size};
 
-        const TARGET: usize = 40;
+        // The user bubble of turn 40: three rows per synthetic turn.
+        const TARGET: usize = 120;
         let timeline = synthetic_markdown_timeline(240);
         let (workspace_store, window_state, _) = seed_chat(cx, timeline);
         let (view, cx) = cx
@@ -5181,7 +5243,7 @@ mod tests {
         });
         assert_eq!(
             view.read_with(cx, |chat, _| chat.resident_markdown_state_count()),
-            48
+            36
         );
         let list_state = view.read_with(cx, |chat, _| chat.list_state.clone());
         assert!(!view.read_with(cx, |chat, _| {
@@ -5224,7 +5286,7 @@ mod tests {
             resident <= 96,
             "old-turn window retained {resident} MarkdownStates; expected at most 96"
         );
-        assert_eq!(resident, 78);
+        assert_eq!(resident, 64);
         let scroll_top = list_state.logical_scroll_top();
         assert_eq!(scroll_top.item_ix, TARGET);
         assert_eq!(
@@ -5245,7 +5307,7 @@ mod tests {
 
         view.read_with(cx, |chat, _| {
             assert!(!chat.has_resident_markdown_state("large"));
-            assert!(!chat.markdown_remeasured_turns.contains(&0));
+            assert!(!chat.markdown_remeasured_rows.contains(&0));
         });
         cx.run_until_parked();
         view.read_with(cx, |chat, cx| {
@@ -5259,7 +5321,7 @@ mod tests {
                 .read(cx)
                 .rendered_text();
             assert!(rendered.contains("async content"));
-            assert!(chat.markdown_remeasured_turns.contains(&0));
+            assert!(chat.markdown_remeasured_rows.contains(&0));
         });
     }
 
@@ -5333,6 +5395,46 @@ mod tests {
         });
     }
 
+    /// A single running turn holding hundreds of interim messages opens
+    /// with the Markdown of the rows near the tail parsed, not the whole
+    /// turn's.
+    #[gpui::test]
+    fn a_long_running_turn_parses_only_the_markdown_near_the_viewport(cx: &mut TestAppContext) {
+        use gpui::px;
+        let mut timeline = Timeline::default();
+        timeline.turn_running = true;
+        timeline.turns = vec![TurnMeta {
+            running: true,
+            ..Default::default()
+        }];
+        timeline.entries.push(entry("user", user_item("go")));
+        for step in 0..150 {
+            timeline.entries.push(command(&format!("cmd-{step}")));
+            timeline
+                .entries
+                .push(entry(&format!("note-{step}"), assistant("interim note")));
+        }
+        let (store, window_state, _) = seed_chat(cx, timeline);
+        let (view, cx) =
+            cx.add_window_view(|window, cx| ChatView::new(store, window_state, window, cx));
+        cx.simulate_resize(gpui::size(px(393.), px(852.)));
+        cx.update(|window, cx| {
+            let _ = window.draw(cx);
+        });
+        cx.run_until_parked();
+        let (rows, resident) = view.read_with(cx, |chat, _| {
+            (chat.rows.len(), chat.resident_markdown_state_count())
+        });
+        assert_eq!(rows, 1 + 2 * 150);
+        assert!(
+            resident < 60,
+            "{resident} Markdown documents resident for a 150-note turn; the rows near the tail suffice"
+        );
+        assert!(view.read_with(cx, |chat, _| chat.has_resident_markdown_state("note-149")));
+        assert!(!view.read_with(cx, |chat, _| chat.has_resident_markdown_state("note-0")));
+        assert!(cx.debug_bounds("timeline-row-300").is_some());
+    }
+
     #[gpui::test]
     fn small_markdown_is_resident_synchronously(cx: &mut TestAppContext) {
         let text = "small **streaming** reply";
@@ -5543,6 +5645,18 @@ This begins after the hard break."#;
         EntryContent::Item(ItemContent::Reasoning { text: text.into() })
     }
 
+    fn command(id: &str) -> Arc<TimelineEntry> {
+        entry(
+            id,
+            EntryContent::Item(ItemContent::CommandExecution {
+                command: id.to_string(),
+                output: String::new(),
+                exit_code: Some(0),
+                status: agent::ItemStatus::Completed,
+            }),
+        )
+    }
+
     fn at_turn(mut entry: Arc<TimelineEntry>, turn: usize) -> Arc<TimelineEntry> {
         Arc::make_mut(&mut entry).turn = turn;
         entry
diff --git a/crates/ui/src/chat/model.rs b/crates/ui/src/chat/model.rs
index 7c4bfbcf..c9d3f9e5 100644
--- a/crates/ui/src/chat/model.rs
+++ b/crates/ui/src/chat/model.rs
@@ -1,5 +1,5 @@
 use std::borrow::Cow;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::hash::{DefaultHasher, Hash as _, Hasher as _};
 use std::ops::Range;
 use std::path::Path;
@@ -11,8 +11,8 @@ use tcode_core::session::{
     EntryContent, SteeringStatus, TimelineEntry, TurnMeta, TurnTiming, parse_orchestrate_callback,
 };
 
-pub(crate) type TurnRenderArgs<'a> = (
-    usize,
+pub(crate) type RowRenderArgs<'a> = (
+    &'a TimelineRow,
     &'a TurnMeta,
     &'a Path,
     &'a [Arc<TimelineEntry>],
@@ -36,6 +36,10 @@ pub(crate) enum Segment<'a> {
 #[derive(Debug)]
 pub(crate) struct SegmentedEntries<'a> {
     pub(crate) flow: Vec<Segment<'a>>,
+    /// The entry indices each segment of `flow` spans, in `entries` order.
+    /// A run's range covers the pending steers it skipped over, so a row that
+    /// re-segments its own range folds to the same segment.
+    pub(crate) ranges: Vec<Range<usize>>,
     pub(crate) pending_steers: Vec<&'a TimelineEntry>,
 }
 
@@ -105,16 +109,23 @@ pub(crate) fn segment_entries<'a>(
     turn_running: bool,
 ) -> SegmentedEntries<'a> {
     let mut segments = Vec::new();
+    let mut ranges = Vec::new();
     let mut activities = Vec::new();
+    let mut run_start = None;
     let mut pending_steers = Vec::new();
     let flush_activities = |segments: &mut Vec<Segment<'a>>,
-                            activities: &mut Vec<&'a TimelineEntry>| {
+                            ranges: &mut Vec<Range<usize>>,
+                            activities: &mut Vec<&'a TimelineEntry>,
+                            run_start: &mut Option<usize>,
+                            end: usize| {
         if !activities.is_empty() {
             segments.push(Segment::ActivityRun(std::mem::take(activities)));
+            ranges.push(run_start.take().expect("an activity run has a start")..end);
         }
+        *run_start = None;
     };
 
-    for entry in entries {
+    for (index, entry) in entries.iter().enumerate() {
         let entry = entry.as_ref();
         if turn_running
             && matches!(
@@ -128,13 +139,23 @@ pub(crate) fn segment_entries<'a>(
             pending_steers.push(entry);
             continue;
         }
+        let mut single = |segments: &mut Vec<Segment<'a>>,
+                          ranges: &mut Vec<Range<usize>>,
+                          segment: Segment<'a>| {
+            flush_activities(segments, ranges, &mut activities, &mut run_start, index);
+            segments.push(segment);
+            ranges.push(index..index + 1);
+        };
         match &entry.content {
             EntryContent::Item(ItemContent::CommandExecution { .. })
             | EntryContent::Item(ItemContent::ToolCall { .. })
             | EntryContent::Item(ItemContent::Subagent { .. })
             | EntryContent::Item(ItemContent::WebSearch { .. })
             | EntryContent::Item(ItemContent::Other { .. })
-            | EntryContent::Item(ItemContent::FileChange { .. }) => activities.push(entry),
+            | EntryContent::Item(ItemContent::FileChange { .. }) => {
+                run_start.get_or_insert(index);
+                activities.push(entry);
+            }
             EntryContent::Item(ItemContent::Reasoning { .. }) => {
                 if activities.last().is_some_and(|previous| {
                     matches!(
@@ -145,41 +166,46 @@ pub(crate) fn segment_entries<'a>(
                 }) {
                     activities.pop();
                 }
+                run_start.get_or_insert(index);
                 activities.push(entry);
             }
             EntryContent::Item(ItemContent::UserMessage { .. }) | EntryContent::Steer { .. } => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::User(entry));
+                single(&mut segments, &mut ranges, Segment::User(entry));
             }
             EntryContent::ProviderRelay { .. } => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::Relay(entry));
+                single(&mut segments, &mut ranges, Segment::Relay(entry));
             }
             EntryContent::ModelChanged { .. } => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::ModelChange(entry));
+                single(&mut segments, &mut ranges, Segment::ModelChange(entry));
             }
             EntryContent::ContextCompacted(_) => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::ContextCompacted(entry));
+                single(&mut segments, &mut ranges, Segment::ContextCompacted(entry));
             }
             EntryContent::ContextWindowChanged { .. } => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::ContextWindowChanged(entry));
+                single(
+                    &mut segments,
+                    &mut ranges,
+                    Segment::ContextWindowChanged(entry),
+                );
             }
             EntryContent::Item(ItemContent::AssistantMessage { .. }) => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::Assistant(entry));
+                single(&mut segments, &mut ranges, Segment::Assistant(entry));
             }
             EntryContent::Error { .. } | EntryContent::ProviderStartError { .. } => {
-                flush_activities(&mut segments, &mut activities);
-                segments.push(Segment::Error(entry));
+                single(&mut segments, &mut ranges, Segment::Error(entry));
             }
         }
     }
-    flush_activities(&mut segments, &mut activities);
+    flush_activities(
+        &mut segments,
+        &mut ranges,
+        &mut activities,
+        &mut run_start,
+        entries.len(),
+    );
     SegmentedEntries {
         flow: segments,
+        ranges,
         pending_steers,
     }
 }
@@ -739,11 +765,27 @@ pub(crate) fn plain_text_as_markdown(text: &str) -> String {
     markdown
 }
 
-/// Cached indexing and cheap height-affecting identity for one virtualized turn.
+/// One virtualized timeline row: a segment of a turn, plus the turn's trailer
+/// (plan card, changed files, liveness, pending steers) on its last row.
+///
+/// The list virtualizes segments rather than turns so a turn with hundreds of
+/// tool runs and interim messages costs the rows on screen, not the whole
+/// turn, every frame. A turn without segments still owns one empty row for
+/// its trailer.
 #[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) struct TurnListItem {
+pub(crate) struct TimelineRow {
+    pub(crate) turn: usize,
     pub(crate) entry_range: Range<usize>,
     pub(crate) entry_count: usize,
+    pub(crate) first_in_turn: bool,
+    pub(crate) last_in_turn: bool,
+    /// The turn's live work log: the run that opens on its own while the
+    /// turn runs ([`live_activity_segment`]).
+    pub(crate) live_activity: bool,
+    /// The turn's final assistant text, the one that carries an action row.
+    pub(crate) last_assistant: bool,
+    /// Identity of the row's first entry: a run grows at its end, so the
+    /// row stays itself while it streams.
     pub(crate) identity: u64,
     /// Identity of the newest entry alone. A history page can only add
     /// entries before a partially loaded turn, so this survives completion
@@ -752,6 +794,27 @@ pub(crate) struct TurnListItem {
     pub(crate) content: u64,
 }
 
+/// The rows of `turn` within `rows`, which are sorted by turn.
+pub(crate) fn rows_of_turn(rows: &[TimelineRow], turn: usize) -> Range<usize> {
+    let start = rows.partition_point(|row| row.turn < turn);
+    let end = start + rows[start..].partition_point(|row| row.turn == turn);
+    start..end
+}
+
+/// The row that renders the entry at `entry_index`. An entry no segment
+/// covers (a pending steer between two runs) renders in its turn's trailer,
+/// the turn's last row.
+pub(crate) fn row_of_entry(rows: &[TimelineRow], entry_index: usize, turn: usize) -> Option<usize> {
+    let candidate = rows.partition_point(|row| row.entry_range.end <= entry_index);
+    if rows
+        .get(candidate)
+        .is_some_and(|row| row.entry_range.contains(&entry_index))
+    {
+        return Some(candidate);
+    }
+    rows_of_turn(rows, turn).last()
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct TurnIndexMeta {
     start_ts: Option<u64>,
@@ -823,7 +886,7 @@ pub(crate) struct TurnIndexCache {
 impl TurnIndexCache {
     pub(crate) fn sync(
         &mut self,
-        items: &mut Vec<TurnListItem>,
+        items: &mut Vec<TimelineRow>,
         turns: &[TurnMeta],
         entries: &[Arc<TimelineEntry>],
         proposed_plan: Option<(usize, &str, &str)>,
@@ -831,7 +894,7 @@ impl TurnIndexCache {
         continuity: TimelineContinuity,
     ) -> ListSync {
         let reset = continuity == TimelineContinuity::NewSession;
-        let item_count = turns
+        let turn_count = turns
             .len()
             .max(entries.last().map_or(0, |entry| entry.turn + 1));
         let entry_divergence = self
@@ -863,7 +926,7 @@ impl TurnIndexCache {
             .zip(turns)
             .position(|(old, new)| !old.matches(new))
             .unwrap_or(self.turns.len().min(turns.len()));
-        let mut reindex_from = if must_reset { 0 } else { item_count };
+        let mut reindex_from = if must_reset { 0 } else { turn_count };
         if !must_reset {
             if entry_divergence < entries.len() {
                 reindex_from = reindex_from.min(entries[entry_divergence].turn);
@@ -877,28 +940,30 @@ impl TurnIndexCache {
         }
 
         let suffix = if reindex_from == 0 {
-            index_turns(turns, entries, proposed_plan, expanded)
+            index_rows(turns, entries, proposed_plan, expanded)
         } else {
-            if reindex_from < item_count {
-                index_turns_from(turns, entries, proposed_plan, expanded, reindex_from)
+            if reindex_from < turn_count {
+                index_rows_from(turns, entries, proposed_plan, expanded, reindex_from)
             } else {
                 Vec::new()
             }
         };
+        // Rows of the turns before `reindex_from` are kept as they are.
+        let kept = rows_of_turn(items, reindex_from).start;
+        let item_count = kept + suffix.len();
         let sync = list_sync_with(items, item_count, continuity, |index| {
-            if index < reindex_from {
+            if index < kept {
                 &items[index]
             } else {
-                &suffix[index - reindex_from]
+                &suffix[index - kept]
             }
         });
-        items.truncate(reindex_from);
+        items.truncate(kept);
         items.extend(suffix);
-        items.truncate(item_count);
 
         #[cfg(test)]
         {
-            self.reindexed_turns = item_count.saturating_sub(reindex_from);
+            self.reindexed_turns = turn_count.saturating_sub(reindex_from);
         }
         self.entries.truncate(entry_divergence);
         self.entries
@@ -925,110 +990,140 @@ impl TurnIndexCache {
     }
 }
 
-/// Mutation to apply to the persistent [`ListState`] after a timeline sync.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) enum ListSync {
-    None,
-    Prepend {
-        count: usize,
-        remeasure: Vec<usize>,
-    },
-    Reset {
-        count: usize,
-    },
-    Incremental {
-        append: Option<Range<usize>>,
-        remeasure: Vec<usize>,
-    },
-}
-
-/// Build contiguous entry ranges and fingerprints for turn-level list items.
+/// Build the segment rows and their fingerprints for every turn.
 ///
 /// Timeline entries are chronological, so all entries for a turn are adjacent.
 /// The max entry turn keeps a temporary orphan bucket renderable if a provider
 /// ever exposes an entry before its corresponding `TurnMeta`.
-pub(crate) fn index_turns(
+pub(crate) fn index_rows(
     turns: &[TurnMeta],
     entries: &[Arc<TimelineEntry>],
     proposed_plan: Option<(usize, &str, &str)>,
     expanded: &HashSet<String>,
-) -> Vec<TurnListItem> {
-    index_turns_from(turns, entries, proposed_plan, expanded, 0)
+) -> Vec<TimelineRow> {
+    index_rows_from(turns, entries, proposed_plan, expanded, 0)
 }
 
-fn index_turns_from(
+fn index_rows_from(
     turns: &[TurnMeta],
     entries: &[Arc<TimelineEntry>],
     proposed_plan: Option<(usize, &str, &str)>,
     expanded: &HashSet<String>,
     first_turn: usize,
-) -> Vec<TurnListItem> {
+) -> Vec<TimelineRow> {
     debug_assert!(entries.windows(2).all(|pair| pair[0].turn <= pair[1].turn));
 
-    let item_count = turns
+    let turn_count = turns
         .len()
         .max(entries.last().map_or(0, |entry| entry.turn + 1));
     let first_entry = entries.partition_point(|entry| entry.turn < first_turn);
-    let mut ranges = vec![entries.len()..entries.len(); item_count.saturating_sub(first_turn)];
+    let mut turn_ranges = vec![entries.len()..entries.len(); turn_count.saturating_sub(first_turn)];
     for (index, entry) in entries.iter().enumerate().skip(first_entry) {
-        let range = &mut ranges[entry.turn - first_turn];
+        let range = &mut turn_ranges[entry.turn - first_turn];
         if range.start == entries.len() {
             range.start = index;
         }
         range.end = index + 1;
     }
 
-    ranges
-        .into_iter()
-        .enumerate()
-        .map(|(offset, entry_range)| {
-            let index = first_turn + offset;
+    let mut rows = Vec::new();
+    for (offset, turn_range) in turn_ranges.into_iter().enumerate() {
+        let index = first_turn + offset;
+        let turn = turns.get(index);
+        let running = turn.is_some_and(|turn| turn.running);
+        let segmented = segment_entries(&entries[turn_range.clone()], running);
+        let live_activity = live_activity_segment(&segmented.flow, running);
+        let last_assistant = segmented
+            .flow
+            .iter()
+            .rposition(|segment| matches!(segment, Segment::Assistant(_)));
+        let mut segment_ranges = segmented
+            .ranges
+            .iter()
+            .map(|range| turn_range.start + range.start..turn_range.start + range.end)
+            .collect::<Vec<_>>();
+        if segment_ranges.is_empty() {
+            segment_ranges.push(turn_range.end..turn_range.end);
+        }
+        let last_segment = segment_ranges.len() - 1;
+        for (segment, entry_range) in segment_ranges.into_iter().enumerate() {
+            let first_in_turn = segment == 0;
+            let last_in_turn = segment == last_segment;
+            let live = live_activity == Some(segment);
+            let final_assistant = last_assistant == Some(segment);
             let mut identity = DefaultHasher::new();
             let mut tail_identity = DefaultHasher::new();
             let mut content = DefaultHasher::new();
             if let Some(entry) = entries[entry_range.clone()].last() {
                 entry.id.hash(&mut tail_identity);
             }
-            for entry in &entries[entry_range.clone()] {
+            if let Some(entry) = entries[entry_range.clone()].first() {
                 entry.id.hash(&mut identity);
+            }
+            for entry in &entries[entry_range.clone()] {
                 std::mem::discriminant(&entry.content).hash(&mut content);
                 entry.ts.hash(&mut content);
                 hash_entry_shape(&entry.content, &mut content);
                 // A disclosure row (orchestrate context / callback) grows a tall
                 // scroll card when expanded, so its toggle state must change the
-                // turn fingerprint or the list keeps the collapsed measurement.
+                // row fingerprint or the list keeps the collapsed measurement.
                 if let Some(key) = disclosure_key(&entry.content, &entry.id) {
                     expanded.contains(&key).hash(&mut content);
                 }
             }
-            if let Some(turn) = turns.get(index) {
-                turn.start_ts.hash(&mut content);
-                turn.end_ts.hash(&mut content);
-                turn.running.hash(&mut content);
-                // The finished bottom row renders the turn's breakdown.
-                turn.timing.hash(&mut content);
-                turn.served_model.hash(&mut content);
-                turn.cost_usd.map(f64::to_bits).hash(&mut content);
-                turn.status
-                    .as_ref()
-                    .map(std::mem::discriminant)
-                    .hash(&mut content);
+            if entry_range.is_empty() {
+                // An empty turn's only row has no entry to be known by.
+                ("empty-turn", index).hash(&mut identity);
             }
-            if let Some((turn, item_id, markdown)) = proposed_plan
-                && turn == index
-            {
-                item_id.hash(&mut identity);
-                markdown.len().hash(&mut content);
+            (first_in_turn, last_in_turn, live, final_assistant).hash(&mut content);
+            running.hash(&mut content);
+            if last_in_turn {
+                // The trailer: pending steers float here while the turn
+                // runs, and the finished turn renders its breakdown.
+                for entry in &entries[turn_range.clone()] {
+                    if let EntryContent::Steer { text, status, .. } = &entry.content {
+                        entry.id.hash(&mut content);
+                        text.len().hash(&mut content);
+                        status.hash(&mut content);
+                    }
+                }
+                entries[turn_range.clone()]
+                    .last()
+                    .and_then(|entry| entry.ts)
+                    .hash(&mut content);
+                if let Some(turn) = turn {
+                    turn.start_ts.hash(&mut content);
+                    turn.end_ts.hash(&mut content);
+                    turn.timing.hash(&mut content);
+                    turn.served_model.hash(&mut content);
+                    turn.cost_usd.map(f64::to_bits).hash(&mut content);
+                    turn.status
+                        .as_ref()
+                        .map(std::mem::discriminant)
+                        .hash(&mut content);
+                }
+                if let Some((turn, item_id, markdown)) = proposed_plan
+                    && turn == index
+                {
+                    item_id.hash(&mut content);
+                    markdown.len().hash(&mut content);
+                }
             }
-            TurnListItem {
+            rows.push(TimelineRow {
+                turn: index,
                 entry_count: entry_range.len(),
                 entry_range,
+                first_in_turn,
+                last_in_turn,
+                live_activity: live,
+                last_assistant: final_assistant,
                 identity: identity.finish(),
                 tail_identity: tail_identity.finish(),
                 content: content.finish(),
-            }
-        })
-        .collect()
+            });
+        }
+    }
+    rows
 }
 
 /// The per-entry expansion key for a user message that renders as a disclosure
@@ -1171,75 +1266,143 @@ fn hash_entry_shape(content: &EntryContent, hash: &mut DefaultHasher) {
     }
 }
 
+/// Mutation to apply to the persistent [`ListState`] after a timeline sync.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum ListSync {
+    None,
+    /// Rows arrived above every existing row: an earlier history page, or
+    /// the records that complete the partial first turn.
+    Prepend {
+        count: usize,
+        remeasure: Vec<usize>,
+    },
+    Reset {
+        count: usize,
+    },
+    /// Rows inserted or replaced in place. `splices` are ranges of the old
+    /// rows and the number of new rows standing in for each, ascending, so
+    /// they apply back to front; `remeasure` indexes the new rows.
+    Incremental {
+        splices: Vec<(Range<usize>, usize)>,
+        remeasure: Vec<usize>,
+    },
+}
+
 #[cfg(test)]
 pub(crate) fn list_sync(
-    old: &[TurnListItem],
-    new: &[TurnListItem],
+    old: &[TimelineRow],
+    new: &[TimelineRow],
     continuity: TimelineContinuity,
 ) -> ListSync {
     list_sync_with(old, new.len(), continuity, |index| &new[index])
 }
 
+/// Align the old rows with the new ones by identity, in order.
+///
+/// Rows keep their measured height wherever their identity survives; a row
+/// whose content changed is remeasured in place. Rows that appear are
+/// spliced in where they stand, which is how the list keeps the reader's
+/// anchor across a steer landing mid-turn or a page arriving above. Rows
+/// that vanish or trade places reset the list, except inside the partial
+/// first turn, whose entries a page may merge or shift.
 fn list_sync_with<'a>(
-    old: &[TurnListItem],
+    old: &[TimelineRow],
     new_len: usize,
     continuity: TimelineContinuity,
-    new_at: impl Fn(usize) -> &'a TurnListItem,
+    new_at: impl Fn(usize) -> &'a TimelineRow,
 ) -> ListSync {
-    let session_changed = continuity == TimelineContinuity::NewSession;
+    if continuity == TimelineContinuity::NewSession {
+        return ListSync::Reset { count: new_len };
+    }
     let partial_first_turn = continuity == TimelineContinuity::PartialFirstTurn;
-    if !session_changed && !old.is_empty() && new_len > old.len() {
-        let count = new_len - old.len();
-        // Earlier history keeps the newest loaded turn, possibly completing
-        // it when it was the partial first turn. A window folded without its
-        // earlier context can also shift entries between other turns; those
-        // rows are remeasured, never reset, so the reading anchor survives
-        // every page.
-        let continues = |index: usize| {
-            let (old, new) = (&old[index], new_at(index + count));
-            old.entry_count > 0
-                && (old.identity == new.identity
-                    || old.tail_identity == new.tail_identity && new.entry_count >= old.entry_count)
-        };
-        if continues(old.len() - 1) {
-            let remeasure = (0..old.len())
-                .filter_map(|index| {
-                    let (old, new) = (&old[index], new_at(index + count));
-                    (old.identity != new.identity || old.content != new.content)
-                        .then_some(index + count)
-                })
-                .collect();
-            return ListSync::Prepend { count, remeasure };
+    let mut old_pos = HashMap::with_capacity(old.len());
+    for (index, row) in old.iter().enumerate() {
+        if old_pos.insert(row.identity, index).is_some() {
+            return ListSync::Reset { count: new_len };
         }
     }
-    let common = old.len().min(new_len);
-    let replaced = (usize::from(partial_first_turn)..common).any(|index| {
-        let old = &old[index];
-        let new = new_at(index);
-        new.entry_count < old.entry_count
-            || (new.entry_count == old.entry_count && new.identity != old.identity)
-    });
-    if session_changed || new_len < old.len() || replaced {
-        return ListSync::Reset { count: new_len };
+    let mut new_pos = HashMap::with_capacity(new_len);
+    for index in 0..new_len {
+        if new_pos.insert(new_at(index).identity, index).is_some() {
+            return ListSync::Reset { count: new_len };
+        }
     }
+    // The partial first turn's row keeps its place when a page completes
+    // it: its newest entry is still its newest entry.
+    let completes = |old_row: &TimelineRow, new_row: &TimelineRow| {
+        partial_first_turn
+            && old_row.turn == 0
+            && old_row.entry_count > 0
+            && old_row.tail_identity == new_row.tail_identity
+            && new_row.entry_count >= old_row.entry_count
+    };
 
-    let append = (new_len > old.len()).then_some(old.len()..new_len);
-    let mut remeasure = (0..common)
-        .filter(|&index| {
-            old[index].entry_count != new_at(index).entry_count
-                || old[index].identity != new_at(index).identity
-                || old[index].content != new_at(index).content
-        })
-        .collect::<Vec<_>>();
-    // The former last item gains an inter-turn gap when a new turn appears.
-    if append.is_some() && !old.is_empty() && !remeasure.contains(&(old.len() - 1)) {
-        remeasure.push(old.len() - 1);
+    let mut splices: Vec<(Range<usize>, usize)> = Vec::new();
+    let mut remeasure = Vec::new();
+    let (mut i, mut j) = (0, 0);
+    while i < old.len() || j < new_len {
+        if i < old.len() && j < new_len {
+            let (old_row, new_row) = (&old[i], new_at(j));
+            if old_row.identity == new_row.identity {
+                if old_row.content != new_row.content || old_row.entry_count != new_row.entry_count
+                {
+                    remeasure.push(j);
+                }
+                i += 1;
+                j += 1;
+                continue;
+            }
+        }
+        let old_survives = i < old.len() && new_pos.contains_key(&old[i].identity);
+        let new_existed = j < new_len && old_pos.contains_key(&new_at(j).identity);
+        if i < old.len() && !old_survives {
+            // The completed first-turn row may sit below rows the page
+            // added above it.
+            if let Some(k) = (j..new_len).find(|&k| completes(&old[i], new_at(k))) {
+                if k > j {
+                    splices.push((i..i, k - j));
+                }
+                remeasure.push(k);
+                i += 1;
+                j = k + 1;
+                continue;
+            }
+            // An empty turn's row goes when the turn gains entries; the
+            // partial first turn's rows may merge as their records arrive.
+            let replaceable =
+                |row: &TimelineRow| row.entry_count == 0 || partial_first_turn && row.turn == 0;
+            if !replaceable(&old[i]) {
+                return ListSync::Reset { count: new_len };
+            }
+            let start = i;
+            while i < old.len() && !new_pos.contains_key(&old[i].identity) && replaceable(&old[i]) {
+                i += 1;
+            }
+            let inserted = j;
+            while j < new_len && !old_pos.contains_key(&new_at(j).identity) {
+                j += 1;
+            }
+            splices.push((start..i, j - inserted));
+            remeasure.extend(inserted..j);
+        } else if j < new_len && !new_existed {
+            let inserted = j;
+            while j < new_len && !old_pos.contains_key(&new_at(j).identity) {
+                j += 1;
+            }
+            splices.push((i..i, j - inserted));
+        } else {
+            // Both rows live on elsewhere: an order change.
+            return ListSync::Reset { count: new_len };
+        }
     }
 
-    if append.is_none() && remeasure.is_empty() {
-        ListSync::None
-    } else {
-        ListSync::Incremental { append, remeasure }
+    match splices.as_slice() {
+        [] if remeasure.is_empty() => ListSync::None,
+        [(range, count)] if range == &(0..0) && !old.is_empty() => ListSync::Prepend {
+            count: *count,
+            remeasure,
+        },
+        _ => ListSync::Incremental { splices, remeasure },
     }
 }
 
@@ -1271,23 +1434,24 @@ mod tests {
                 text: "Second response".into(),
             }),
         });
-        let old = index_turns(
+        let old = index_rows(
             &vec![TurnMeta::default(); 2],
             std::slice::from_ref(&first),
             None,
             &HashSet::new(),
         );
-        let new = index_turns(
+        let new = index_rows(
             &vec![TurnMeta::default(); 3],
             &[first, second],
             None,
             &HashSet::new(),
         );
+        // The empty turn's row gives way to the entry and the new empty turn.
         assert_eq!(
             list_sync(&old, &new, TimelineContinuity::Complete),
             ListSync::Incremental {
-                append: Some(2..3),
-                remeasure: vec![1]
+                splices: vec![(1..2, 2)],
+                remeasure: vec![1, 2]
             }
         );
     }
@@ -1298,7 +1462,7 @@ mod tests {
     #[test]
     fn history_completing_the_only_partial_turn_prepends_instead_of_resetting() {
         let index = |entries: &[Arc<TimelineEntry>], turns: usize| {
-            index_turns(
+            index_rows(
                 &vec![TurnMeta::default(); turns],
                 entries,
                 None,
@@ -1319,8 +1483,8 @@ mod tests {
                 TimelineContinuity::PartialFirstTurn
             ),
             ListSync::Prepend {
-                count: 1,
-                remeasure: vec![1]
+                count: 3,
+                remeasure: vec![3]
             }
         );
 
@@ -1337,20 +1501,20 @@ mod tests {
         );
         let refolded = [
             at_turn(entry("z", user_item("z")), 0),
-            at_turn(entry("shifted", reasoning("moved")), 0),
-            at_turn(entry("a", assistant("a")), 1),
-            at_turn(entry("b", assistant("b")), 2),
-            at_turn(entry("c", assistant("c")), 3),
+            at_turn(entry("a", assistant("a")), 0),
+            at_turn(entry("b", assistant("b")), 1),
+            at_turn(entry("shifted", reasoning("moved")), 2),
+            at_turn(entry("c", assistant("c")), 2),
         ];
         assert_eq!(
             list_sync(
                 &old,
-                &index(&refolded, 4),
+                &index(&refolded, 3),
                 TimelineContinuity::PartialFirstTurn
             ),
             ListSync::Prepend {
                 count: 1,
-                remeasure: vec![2]
+                remeasure: vec![1, 2, 3, 4]
             }
         );
 
@@ -1377,7 +1541,7 @@ mod tests {
                 TimelineContinuity::PartialFirstTurn
             ),
             ListSync::Incremental {
-                append: None,
+                splices: vec![(1..2, 0)],
                 remeasure: vec![0]
             }
         );
@@ -1552,51 +1716,69 @@ mod tests {
             entry("user-0", user_item("go")),
             entry("assistant-0", assistant("working")),
         ];
-        let initial = index_turns(&turns, &entries, None, &expanded);
-        assert_eq!(initial.len(), 1);
-        assert_eq!(initial[0].entry_range, 0..2);
-
-        // Another entry joins the current turn: identity stays at item index 0,
-        // but its variable height must be measured again.
+        let initial = index_rows(&turns, &entries, None, &expanded);
+        // One row per segment: the user bubble and the assistant message.
+        assert_eq!(initial.len(), 2);
+        assert_eq!(initial[0].entry_range, 0..1);
+        assert_eq!(initial[1].entry_range, 1..2);
+        assert!(initial[0].first_in_turn && !initial[0].last_in_turn);
+        assert!(!initial[1].first_in_turn && initial[1].last_in_turn);
+
+        // A command joins the current turn as a new row; the former last
+        // row hands over the turn's trailer, so its height is measured again.
         entries.push(command("command-0"));
-        let current_turn_append = index_turns(&turns, &entries, None, &expanded);
-        assert_eq!(current_turn_append[0].entry_range, 0..3);
+        let current_turn_append = index_rows(&turns, &entries, None, &expanded);
+        assert_eq!(current_turn_append[2].entry_range, 2..3);
         assert_eq!(
             list_sync(&initial, &current_turn_append, TimelineContinuity::Complete),
             ListSync::Incremental {
-                append: None,
-                remeasure: vec![0],
+                splices: vec![(2..2, 1)],
+                remeasure: vec![1],
             }
         );
 
-        // A new turn adds exactly one list item. The former tail is also
-        // remeasured because it gains the visual inter-turn gap.
-        let turns = vec![TurnMeta::default(), TurnMeta::default()];
-        entries.push(at_turn(entry("user-1", user_item("next")), 1));
-        let new_turn = index_turns(&turns, &entries, None, &expanded);
-        assert_eq!(new_turn[0].entry_range, 0..3);
-        assert_eq!(new_turn[1].entry_range, 3..4);
+        // A second command joins the same run: the row grows in place.
+        entries.push(command("command-1"));
+        let run_grows = index_rows(&turns, &entries, None, &expanded);
+        assert_eq!(run_grows.len(), 3);
+        assert_eq!(run_grows[2].entry_range, 2..4);
         assert_eq!(
             list_sync(
                 &current_turn_append,
-                &new_turn,
+                &run_grows,
                 TimelineContinuity::Complete
             ),
             ListSync::Incremental {
-                append: Some(1..2),
-                remeasure: vec![0],
+                splices: vec![],
+                remeasure: vec![2],
+            }
+        );
+
+        // A new turn adds exactly one row; the splice's neighbour above is
+        // remeasured by the view for the inter-turn gap it gains.
+        let turns = vec![TurnMeta::default(), TurnMeta::default()];
+        entries.push(at_turn(entry("user-1", user_item("next")), 1));
+        let new_turn = index_rows(&turns, &entries, None, &expanded);
+        assert_eq!(new_turn[2].entry_range, 2..4);
+        assert_eq!(new_turn[3].entry_range, 4..5);
+        assert_eq!(new_turn[3].turn, 1);
+        assert_eq!(
+            list_sync(&run_grows, &new_turn, TimelineContinuity::Complete),
+            ListSync::Incremental {
+                splices: vec![(3..3, 1)],
+                remeasure: vec![],
             }
         );
 
         // Conversation truncation cannot leave ListState with stale item indices.
         assert_eq!(
             list_sync(&new_turn, &initial, TimelineContinuity::Complete),
-            ListSync::Reset { count: 1 }
+            ListSync::Reset { count: 2 }
         );
         // Even an equal-shaped replacement must reset when the session changes.
         assert_eq!(
             list_sync(&initial, &initial, TimelineContinuity::NewSession),
-            ListSync::Reset { count: 1 }
+            ListSync::Reset { count: 2 }
         );
     }
 
@@ -1616,7 +1798,7 @@ mod tests {
             &expanded,
             TimelineContinuity::Complete,
         );
-        assert_eq!(indexed, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(indexed, index_rows(&turns, &entries, None, &expanded));
 
         // (a) Append an entry to the last turn.
         entries.push(entry("assistant-0", assistant("working")));
@@ -1628,7 +1810,7 @@ mod tests {
             &expanded,
             TimelineContinuity::Complete,
         );
-        assert_eq!(indexed, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(indexed, index_rows(&turns, &entries, None, &expanded));
 
         // (b) Append a new turn.
         turns.push(TurnMeta::default());
@@ -1641,7 +1823,7 @@ mod tests {
             &expanded,
             TimelineContinuity::Complete,
         );
-        assert_eq!(indexed, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(indexed, index_rows(&turns, &entries, None, &expanded));
 
         // (c) Replace the streaming tail Arc with updated content.
         entries[2] = at_turn(entry("user-1", user_item("next, updated")), 1);
@@ -1653,7 +1835,7 @@ mod tests {
             &expanded,
             TimelineContinuity::Complete,
         );
-        assert_eq!(indexed, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(indexed, index_rows(&turns, &entries, None, &expanded));
 
         // (d) Toggle a disclosure expansion key.
         entries[2] = at_turn(
@@ -1684,7 +1866,7 @@ mod tests {
             &expanded,
             TimelineContinuity::Complete,
         );
-        assert_eq!(indexed, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(indexed, index_rows(&turns, &entries, None, &expanded));
 
         // (e) A session switch resets unrelated cached inputs.
         let switched_turns = vec![TurnMeta::default()];
@@ -1700,7 +1882,7 @@ mod tests {
         );
         assert_eq!(
             indexed,
-            index_turns(&switched_turns, &switched_entries, None, &no_expanded)
+            index_rows(&switched_turns, &switched_entries, None, &no_expanded)
         );
 
         // (f) Rewind/removal takes the full path and remains equivalent.
@@ -1715,7 +1897,7 @@ mod tests {
         );
         assert_eq!(
             indexed,
-            index_turns(&switched_turns, &empty_entries, None, &no_expanded)
+            index_rows(&switched_turns, &empty_entries, None, &no_expanded)
         );
     }
 
@@ -1747,7 +1929,7 @@ mod tests {
             TimelineContinuity::Complete,
         );
 
-        assert_eq!(incremental, index_turns(&turns, &entries, None, &expanded));
+        assert_eq!(incremental, index_rows(&turns, &entries, None, &expanded));
         assert!(
             cache.reindexed_turns() <= 1,
             "tail replacement reindexed {} turns",
@@ -1769,7 +1951,7 @@ mod tests {
                 effort: None,
             }),
         )];
-        let running = index_turns(&turns, &entries, None, &HashSet::new());
+        let running = index_rows(&turns, &entries, None, &HashSet::new());
 
         let mut completed_entries = entries;
         if let EntryContent::Item(ItemContent::Subagent {
@@ -1779,11 +1961,11 @@ mod tests {
             *status = ItemStatus::Completed;
             *summary = Some("Found the event envelope".into());
         }
-        let completed = index_turns(&turns, &completed_entries, None, &HashSet::new());
+        let completed = index_rows(&turns, &completed_entries, None, &HashSet::new());
         assert_eq!(
             list_sync(&running, &completed, TimelineContinuity::Complete),
             ListSync::Incremental {
-                append: None,
+                splices: vec![],
                 remeasure: vec![0],
             }
         );
@@ -1803,19 +1985,19 @@ mod tests {
                 status: ItemStatus::InProgress,
             }),
         )];
-        let running = index_turns(&turns, &entries, None, &HashSet::new());
+        let running = index_rows(&turns, &entries, None, &HashSet::new());
         let mut completed_entries = entries;
         if let EntryContent::Item(ItemContent::FileChange { status, .. }) =
             &mut Arc::make_mut(&mut completed_entries[0]).content
         {
             *status = ItemStatus::Completed;
         }
-        let completed = index_turns(&turns, &completed_entries, None, &HashSet::new());
+        let completed = index_rows(&turns, &completed_entries, None, &HashSet::new());
 
         assert_eq!(
             list_sync(&running, &completed, TimelineContinuity::Complete),
             ListSync::Incremental {
-                append: None,
+                splices: vec![],
                 remeasure: vec![0],
             }
         );
@@ -1938,6 +2120,57 @@ mod tests {
         assert!(matches!(idle.flow[3], Segment::User(entry) if entry.id == "steer-b"));
     }
 
+    /// The conversation that motivated segment rows: one running turn with
+    /// hundreds of tool runs and interim messages. A turn is not a row; each
+    /// of its segments is, so the list can skip the ones off screen.
+    #[test]
+    fn a_long_running_turn_virtualizes_into_one_row_per_segment() {
+        let turns = vec![TurnMeta {
+            running: true,
+            ..Default::default()
+        }];
+        let mut entries = vec![entry("user", user_item("go"))];
+        for step in 0..200 {
+            entries.push(command(&format!("cmd-{step}")));
+            entries.push(entry(&format!("reasoning-{step}"), reasoning("thinking")));
+            entries.push(entry(&format!("note-{step}"), assistant("interim")));
+        }
+        entries.push(command("cmd-final"));
+        let rows = index_rows(&turns, &entries, None, &HashSet::new());
+
+        // The user bubble, then a work log and a note per step, then the
+        // trailing live work log.
+        assert_eq!(rows.len(), 1 + 2 * 200 + 1);
+        assert!(rows.iter().all(|row| row.turn == 0));
+        assert!(rows[0].first_in_turn && !rows[0].last_in_turn);
+        assert_eq!(
+            rows[1].entry_range,
+            1..3,
+            "a run and its reasoning share a row"
+        );
+        let live = rows.iter().filter(|row| row.live_activity).count();
+        let last_assistant = rows.iter().position(|row| row.last_assistant);
+        assert_eq!(live, 1);
+        assert!(rows.last().unwrap().live_activity);
+        assert!(rows.last().unwrap().last_in_turn);
+        assert_eq!(last_assistant, Some(rows.len() - 2));
+        assert_eq!(rows_of_turn(&rows, 0), 0..rows.len());
+        assert_eq!(row_of_entry(&rows, 2, 0), Some(1));
+        assert_eq!(row_of_entry(&rows, 3, 0), Some(2));
+
+        // The next command grows the last row in place, so the rows on
+        // screen keep their measured heights.
+        entries.push(command("cmd-next"));
+        let grown = index_rows(&turns, &entries, None, &HashSet::new());
+        assert_eq!(
+            list_sync(&rows, &grown, TimelineContinuity::Complete),
+            ListSync::Incremental {
+                splices: vec![],
+                remeasure: vec![rows.len() - 1],
+            }
+        );
+    }
+
     #[test]
     fn steer_status_and_reordering_invalidate_the_virtualized_turn_row() {
         let turns = vec![TurnMeta {
@@ -1955,7 +2188,7 @@ mod tests {
             },
         );
         let assistant = entry("assistant", assistant("working"));
-        let before = index_turns(
+        let before = index_rows(
             &turns,
             &[pending.clone(), assistant.clone()],
             None,
@@ -1966,24 +2199,26 @@ mod tests {
         if let EntryContent::Steer { status, .. } = &mut Arc::make_mut(&mut accepted).content {
             *status = SteeringStatus::Accepted;
         }
-        let status_changed = index_turns(
+        let status_changed = index_rows(
             &turns,
             &[accepted.clone(), assistant.clone()],
             None,
             &expanded,
         );
+        // The pending steer floated in the trailer; accepted, it takes its
+        // recorded place as a row above the answer, which keeps its height.
         assert_eq!(
             list_sync(&before, &status_changed, TimelineContinuity::Complete),
-            ListSync::Incremental {
-                append: None,
-                remeasure: vec![0],
+            ListSync::Prepend {
+                count: 1,
+                remeasure: vec![1],
             }
         );
 
-        let reordered = index_turns(&turns, &[assistant, accepted], None, &expanded);
+        let reordered = index_rows(&turns, &[assistant, accepted], None, &expanded);
         assert_eq!(
             list_sync(&status_changed, &reordered, TimelineContinuity::Complete),
-            ListSync::Reset { count: 1 }
+            ListSync::Reset { count: 2 }
         );
     }
 
diff --git a/crates/ui/src/chat/residency.rs b/crates/ui/src/chat/residency.rs
index 23618336..ccc8103e 100644
--- a/crates/ui/src/chat/residency.rs
+++ b/crates/ui/src/chat/residency.rs
@@ -1,29 +1,29 @@
 use std::collections::HashSet;
 use std::ops::Range;
 
-/// Before GPUI reports its exact visible range, eight turns is comfortably
+/// Before GPUI reports its exact visible range, twelve rows is comfortably
 /// more than a typical chat viewport. Its list also pre-measures four viewport
 /// heights, so the wider eviction band keeps warm rows resident without tying
 /// Markdown lifetime to measurement.
-const VIEWPORT_TURN_HINT: usize = 8;
-const BUILD_MARGIN_TURNS: usize = 8;
+const VIEWPORT_ROW_HINT: usize = 12;
+const BUILD_MARGIN_ROWS: usize = 24;
 /// Three build margins prevent back-and-forth scrolling from rebuilding the
 /// same parsed documents at the edge of the warm window.
-const EVICT_MARGIN_TURNS: usize = 24;
-/// The composer-adjacent tail stays ready even while inspecting old turns.
-const TAIL_PIN_TURNS: usize = 2;
+const EVICT_MARGIN_ROWS: usize = 72;
+/// The composer-adjacent tail stays ready even while inspecting old rows.
+const TAIL_PIN_ROWS: usize = 4;
 
 #[derive(Clone, Debug)]
 pub(super) struct MarkdownEntry {
     pub id: String,
-    pub turn: usize,
-    pub turn_running: bool,
+    /// The timeline row that renders this document.
+    pub row: usize,
 }
 
 pub(super) struct ResidencyInput<'a> {
-    pub turn_count: usize,
-    pub visible_turns: Range<usize>,
-    pub one_shot_turn_target: Option<usize>,
+    pub row_count: usize,
+    pub visible_rows: Range<usize>,
+    pub one_shot_row_target: Option<usize>,
     pub entries: &'a [MarkdownEntry],
     pub stream_running: bool,
     pub resident_ids: &'a HashSet<String>,
@@ -31,43 +31,45 @@ pub(super) struct ResidencyInput<'a> {
     pub selection_drag_active: bool,
 }
 
-/// Turn-only superset of entries that can affect [`decide`].
+/// Row-only superset of entries that can affect [`decide`].
 pub(super) struct ResidencyScope {
-    build_turns: Range<usize>,
-    keep_turns: Range<usize>,
+    build_rows: Range<usize>,
+    keep_rows: Range<usize>,
     tail_start: usize,
-    last_turn: Option<usize>,
+    last_row: Option<usize>,
     stream_running: bool,
 }
 
 impl ResidencyScope {
     pub(super) fn new(
-        turn_count: usize,
-        visible_turns: Range<usize>,
-        one_shot_turn_target: Option<usize>,
+        row_count: usize,
+        visible_rows: Range<usize>,
+        one_shot_row_target: Option<usize>,
         stream_running: bool,
     ) -> Self {
-        let visible_turns = one_shot_turn_target
-            .filter(|turn| *turn < turn_count)
-            .map(|turn| turn..(turn + VIEWPORT_TURN_HINT).min(turn_count))
-            .unwrap_or(visible_turns);
+        let visible_rows = one_shot_row_target
+            .filter(|row| *row < row_count)
+            .map(|row| row..(row + VIEWPORT_ROW_HINT).min(row_count))
+            .unwrap_or(visible_rows);
         Self {
-            build_turns: expand_turn_window(visible_turns.clone(), BUILD_MARGIN_TURNS, turn_count),
-            keep_turns: expand_turn_window(visible_turns, EVICT_MARGIN_TURNS, turn_count),
-            tail_start: turn_count.saturating_sub(TAIL_PIN_TURNS),
-            last_turn: turn_count.checked_sub(1),
+            build_rows: expand_row_window(visible_rows.clone(), BUILD_MARGIN_ROWS, row_count),
+            keep_rows: expand_row_window(visible_rows, EVICT_MARGIN_ROWS, row_count),
+            tail_start: row_count.saturating_sub(TAIL_PIN_ROWS),
+            last_row: row_count.checked_sub(1),
             stream_running,
         }
     }
 
-    pub(super) fn includes(&self, turn: usize, turn_running: bool) -> bool {
-        self.keep_turns.contains(&turn) || self.pinned(turn, turn_running)
+    pub(super) fn includes(&self, row: usize) -> bool {
+        self.keep_rows.contains(&row) || self.pinned(row)
     }
 
-    fn pinned(&self, turn: usize, turn_running: bool) -> bool {
-        self.last_turn.is_some() && turn >= self.tail_start
-            || turn_running
-            || self.stream_running && self.last_turn == Some(turn)
+    /// The tail rows stay resident: the composer sits under them, and a
+    /// running turn streams into the last of them. The rest of a running
+    /// turn is ordinary history, however long it grows.
+    fn pinned(&self, row: usize) -> bool {
+        self.last_row.is_some() && row >= self.tail_start
+            || self.stream_running && self.last_row == Some(row)
     }
 }
 
@@ -79,18 +81,16 @@ pub(super) struct ResidencyDecisions {
 
 pub(super) fn decide(input: ResidencyInput<'_>) -> ResidencyDecisions {
     let scope = ResidencyScope::new(
-        input.turn_count,
-        input.visible_turns,
-        input.one_shot_turn_target,
+        input.row_count,
+        input.visible_rows,
+        input.one_shot_row_target,
         input.stream_running,
     );
 
     let build = input
         .entries
         .iter()
-        .filter(|entry| {
-            scope.build_turns.contains(&entry.turn) || scope.pinned(entry.turn, entry.turn_running)
-        })
+        .filter(|entry| scope.build_rows.contains(&entry.row) || scope.pinned(entry.row))
         .map(|entry| entry.id.clone())
         .collect();
     let evict = if input.selection_drag_active {
@@ -99,7 +99,7 @@ pub(super) fn decide(input: ResidencyInput<'_>) -> ResidencyDecisions {
         let keep = input
             .entries
             .iter()
-            .filter(|entry| scope.includes(entry.turn, entry.turn_running))
+            .filter(|entry| scope.includes(entry.row))
             .map(|entry| entry.id.as_str())
             .collect::<HashSet<_>>();
         input
@@ -113,45 +113,45 @@ pub(super) fn decide(input: ResidencyInput<'_>) -> ResidencyDecisions {
     ResidencyDecisions { build, evict }
 }
 
-pub(super) fn tail_turn_window(turn_count: usize) -> Range<usize> {
-    turn_count.saturating_sub(VIEWPORT_TURN_HINT)..turn_count
+pub(super) fn tail_row_window(row_count: usize) -> Range<usize> {
+    row_count.saturating_sub(VIEWPORT_ROW_HINT)..row_count
 }
 
-pub(super) fn viewport_turn_window(scroll_top: usize, turn_count: usize) -> Range<usize> {
-    scroll_top..(scroll_top + VIEWPORT_TURN_HINT).min(turn_count)
+pub(super) fn viewport_row_window(scroll_top: usize, row_count: usize) -> Range<usize> {
+    scroll_top..(scroll_top + VIEWPORT_ROW_HINT).min(row_count)
 }
 
-fn expand_turn_window(window: Range<usize>, margin: usize, turn_count: usize) -> Range<usize> {
-    window.start.min(turn_count).saturating_sub(margin)
+fn expand_row_window(window: Range<usize>, margin: usize, row_count: usize) -> Range<usize> {
+    window.start.min(row_count).saturating_sub(margin)
         ..window
             .end
-            .min(turn_count)
+            .min(row_count)
             .saturating_add(margin)
-            .min(turn_count)
+            .min(row_count)
 }
 
 #[cfg(test)]
 mod tests {
     use super::{
-        MarkdownEntry, ResidencyDecisions, ResidencyInput, ResidencyScope, decide, tail_turn_window,
+        MarkdownEntry, ResidencyDecisions, ResidencyInput, ResidencyScope, decide, tail_row_window,
     };
     use std::collections::HashSet;
 
     #[test]
     fn residency_stays_bounded_across_small_scrolls_and_distant_jumps() {
         let entries = entries(240);
-        let initial = decisions(&entries, tail_turn_window(240), None, &HashSet::new());
-        assert_eq!(initial.build.len(), 48);
+        let initial = decisions(&entries, tail_row_window(240), None, &HashSet::new());
+        assert_eq!(initial.build.len(), 36);
         assert!(initial.evict.is_empty());
         let mut residents = initial.build;
         let shifted = decisions(&entries, 230..238, None, &residents);
         apply(&mut residents, shifted);
-        let back_to_tail = decisions(&entries, tail_turn_window(240), None, &residents);
+        let back_to_tail = decisions(&entries, tail_row_window(240), None, &residents);
         assert!(back_to_tail.build.is_subset(&residents));
         assert!(back_to_tail.evict.is_empty());
-        let jump = decisions(&entries, tail_turn_window(240), Some(40), &residents);
+        let jump = decisions(&entries, tail_row_window(240), Some(40), &residents);
         apply(&mut residents, jump);
-        assert_eq!(residents.len(), 78);
+        assert_eq!(residents.len(), 64);
         assert!(residents.contains("assistant-40"));
         assert!(!residents.contains("assistant-230"));
         assert!(residents.contains("assistant-238"));
@@ -160,38 +160,32 @@ mod tests {
 
     #[test]
     fn running_tail_and_selection_pins_are_honored() {
-        let mut entries = entries(240);
-        for entry in &mut entries {
-            entry.turn_running = entry.turn == 5;
-        }
-        let residents = ["assistant-7".to_string(), "assistant-100".to_string()]
+        let entries = entries(240);
+        let residents = ["assistant-7".to_string(), "assistant-150".to_string()]
             .into_iter()
             .collect();
         let selection_participants = ["assistant-7".to_string()].into_iter().collect();
         let decisions = decide(ResidencyInput {
-            turn_count: 240,
-            visible_turns: 40..48,
-            one_shot_turn_target: None,
+            row_count: 240,
+            visible_rows: 40..48,
+            one_shot_row_target: None,
             entries: &entries,
-            stream_running: false,
+            stream_running: true,
             resident_ids: &residents,
             selection_participants: &selection_participants,
             selection_drag_active: false,
         });
 
-        assert!(decisions.build.contains("assistant-5"));
+        assert!(!decisions.build.contains("assistant-5"));
         assert!(decisions.build.contains("assistant-238"));
         assert!(decisions.build.contains("assistant-239"));
         assert!(!decisions.evict.contains("assistant-7"));
-        assert!(decisions.evict.contains("assistant-100"));
+        assert!(decisions.evict.contains("assistant-150"));
     }
 
     #[test]
-    fn candidate_filter_preserves_decisions_with_distant_running_and_selection() {
-        let mut all_entries = entries(200);
-        for entry in &mut all_entries {
-            entry.turn_running = entry.turn == 5;
-        }
+    fn candidate_filter_preserves_decisions_with_distant_tail_and_selection() {
+        let all_entries = entries(200);
         let residents = [
             "assistant-5".to_string(),
             "assistant-40".to_string(),
@@ -204,14 +198,14 @@ mod tests {
         let scope = ResidencyScope::new(200, 40..48, None, true);
         let filtered_entries = all_entries
             .iter()
-            .filter(|entry| scope.includes(entry.turn, entry.turn_running))
+            .filter(|entry| scope.includes(entry.row))
             .cloned()
             .collect::<Vec<_>>();
         let run = |entries: &[MarkdownEntry]| {
             decide(ResidencyInput {
-                turn_count: 200,
-                visible_turns: 40..48,
-                one_shot_turn_target: None,
+                row_count: 200,
+                visible_rows: 40..48,
+                one_shot_row_target: None,
                 entries,
                 stream_running: true,
                 resident_ids: &residents,
@@ -221,32 +215,30 @@ mod tests {
         };
 
         assert_eq!(run(&filtered_entries), run(&all_entries));
-        assert!(run(&filtered_entries).build.contains("assistant-5"));
-        assert!(!run(&filtered_entries).evict.contains("assistant-100"));
+        assert!(run(&filtered_entries).build.contains("assistant-199"));
+        assert!(run(&filtered_entries).evict.is_empty());
     }
 
-    fn entries(turn_count: usize) -> Vec<MarkdownEntry> {
-        (0..turn_count)
-            .flat_map(|turn| {
-                ["user", "reasoning", "assistant"].map(move |kind| MarkdownEntry {
-                    id: format!("{kind}-{turn}"),
-                    turn,
-                    turn_running: false,
-                })
+    /// One document per row, as a conversation of single-message rows has.
+    fn entries(row_count: usize) -> Vec<MarkdownEntry> {
+        (0..row_count)
+            .map(|row| MarkdownEntry {
+                id: format!("assistant-{row}"),
+                row,
             })
             .collect()
     }
 
     fn decisions(
         entries: &[MarkdownEntry],
-        visible_turns: std::ops::Range<usize>,
-        one_shot_turn_target: Option<usize>,
+        visible_rows: std::ops::Range<usize>,
+        one_shot_row_target: Option<usize>,
         resident_ids: &HashSet<String>,
     ) -> ResidencyDecisions {
         decide(ResidencyInput {
-            turn_count: 240,
-            visible_turns,
-            one_shot_turn_target,
+            row_count: 240,
+            visible_rows,
+            one_shot_row_target,
             entries,
             stream_running: false,
             resident_ids,
diff --git a/crates/ui/src/store/history.rs b/crates/ui/src/store/history.rs
index fbbcc3c2..e7a0ea24 100644
--- a/crates/ui/src/store/history.rs
+++ b/crates/ui/src/store/history.rs
@@ -101,7 +101,7 @@ impl WorkspaceStore {
                         let held = store.session_records.entry(session_id.clone()).or_default();
                         held.splice(0..0, records);
                         store.session_from.insert(session_id.clone(), from);
-                        let mut timeline = Timeline::fold_events(held.iter().cloned());
+                        let mut timeline = Timeline::fold_stored(held.iter());
                         if !store
                             .session_status_replica
                             .as_ref()
diff --git a/crates/ui/src/store/mod.rs b/crates/ui/src/store/mod.rs
index 3f5880ce..31bb9e63 100644
--- a/crates/ui/src/store/mod.rs
+++ b/crates/ui/src/store/mod.rs
@@ -1033,7 +1033,7 @@ impl WorkspaceStore {
                 if self.session_catching_up {
                     return;
                 }
-                let mut timeline = Timeline::fold_events(held.iter().cloned());
+                let mut timeline = Timeline::fold_stored(held.iter());
                 if !self
                     .session_status_replica
                     .as_ref()