Retain compositional line sequences and synchronize worker shutdown

native/src/atom_plan.rs

@@ -318,6 +318,10 @@ impl AtomPlan {
         self.parts.is_empty()
     }
 
+    pub(super) fn char_count(&self) -> u64 {
+        self.parts.measure().chars
+    }
+
     pub(super) fn append(&self, other: &Self) -> Self {
         if self.is_empty() {
             return other.clone();

native/src/layout.rs

@@ -19,6 +19,19 @@ pub(crate) fn atom_plan_work() -> AtomPlanWork {
     atom_plan::work()
 }
 
+#[path = "line_plan.rs"]
+mod line_plan;
+use line_plan::LinePlan;
+pub(crate) use line_plan::LinePlanWork;
+
+pub(crate) fn reset_line_plan_work() {
+    line_plan::reset_work();
+}
+
+pub(crate) fn line_plan_work() -> LinePlanWork {
+    line_plan::work()
+}
+
 fn deserialize_arc_vec<'de, D, T>(deserializer: D) -> Result<Arc<Vec<T>>, D::Error>
 where
     D: Deserializer<'de>,
@@ -1835,14 +1848,18 @@ impl Atom {
 
 #[derive(Debug, Clone)]
 struct Rendered {
-    lines: Vec<Line>,
+    lines: LinePlan,
-    breaks: Vec<AtomProperties>,
 }
 
 impl Rendered {
     fn from_lines(lines: Vec<Line>) -> Self {
-        let breaks = vec![AtomProperties::default(); lines.len().saturating_sub(1)];
+        Self {
-        Self { lines, breaks }
+            lines: LinePlan::from_lines(lines),
+        }
+    }
+
+    fn from_line_plan(lines: LinePlan) -> Self {
+        Self { lines }
     }
 
     fn first_width(&self) -> i64 {
@@ -1850,7 +1867,7 @@ impl Rendered {
     }
 
     fn max_width(&self) -> i64 {
-        self.lines.iter().map(|line| line.width).max().unwrap_or(0)
+        self.lines.max_width()
     }
 
     fn min_content_width(&self, wrap_mode: WrapMode) -> i64 {
@@ -1865,28 +1882,29 @@ impl Rendered {
     }
 
     fn height(&self) -> i64 {
-        self.lines.len() as i64
+        self.lines.height()
     }
 
     fn apply_scroll_window(&mut self, region_id: i64) {
-        for line in &mut self.lines {
+        self.lines = self
-            line.atoms = line.atoms.apply_scroll_window(region_id);
+            .lines
-        }
+            .map_lines(|_, mut line| {
-        for properties in &mut self.breaks {
+                line.atoms = line.atoms.apply_scroll_window(region_id);
-            properties.scroll_window = Some(region_id);
+                line
-        }
+            })
+            .map_breaks(|properties| {
+                properties.scroll_window = Some(region_id);
+            });
     }
 
     fn into_tape(self, style_count: u32) -> LayoutTape {
-        let last_line = self.lines.len().saturating_sub(1);
+        line_plan::record_materialized(self.lines.len(), self.lines.prefix_chars(self.lines.len()));
-        let mut breaks = self.breaks.into_iter();
         LayoutTape {
             style_count,
             lines: self
                 .lines
-                .into_iter()
+                .iter_with_breaks()
-                .enumerate()
+                .map(|(line, break_after)| TapeLine {
-                .map(|(index, line)| TapeLine {
                     width: line.width,
                     atoms: line
                         .atoms
@@ -1908,8 +1926,7 @@ impl Rendered {
                             }
                         })
                         .collect(),
-                    break_after: (index < last_line)
+                    break_after: break_after.cloned(),
-                        .then(|| breaks.next().expect("rendered break invariant")),
                 })
                 .collect(),
         }
@@ -5026,7 +5043,7 @@ fn wrap_rendered_line(line: &Line, max_width: i64, mode: WrapMode) -> Vec<Line>
         return vec![Line::default()];
     }
     if mode == WrapMode::None || line.width <= max_width {
-        return vec![line.clone()];
+        return vec![line_plan::clone_line(line)];
     }
 
     // Wrapping inspects the actual atom boundaries. Composition and decoration
@@ -5112,34 +5129,18 @@ fn wrap_rendered(rendered: Rendered, max_width: i64, mode: WrapMode) -> Rendered
     if mode == WrapMode::None {
         return rendered;
     }
-    let original_breaks = rendered.breaks;
+    let mut lines = LinePlan::default();
-    let mut lines = Vec::new();
+    let mut joining_break = AtomProperties::default();
-    let mut breaks = Vec::new();
+    for (line, break_after) in rendered.lines.iter_with_breaks() {
-    for (line_index, line) in rendered.lines.into_iter().enumerate() {
+        let pieces = LinePlan::from_lines(wrap_rendered_line(line, max_width, mode));
-        for (piece_index, piece) in wrap_rendered_line(&line, max_width, mode)
+        lines = lines.concat_with_break(&pieces, joining_break);
-            .into_iter()
+        joining_break = break_after.cloned().unwrap_or_default();
-            .enumerate()
-        {
-            if !lines.is_empty() {
-                breaks.push(if piece_index == 0 {
-                    original_breaks[line_index - 1].clone()
-                } else {
-                    AtomProperties::default()
-                });
-            }
-            lines.push(piece);
-        }
     }
-    Rendered { lines, breaks }
+    Rendered { lines }
 }
 
-fn vertical_align(
+fn vertical_align(lines: LinePlan, height: usize, align: VerticalAlign, width: i64) -> LinePlan {
-    mut lines: Vec<Line>,
+    let lines = lines.slice(0..lines.len().min(height));
-    height: usize,
-    align: VerticalAlign,
-    width: i64,
-) -> Vec<Line> {
-    lines.truncate(height);
     if lines.len() >= height {
         return lines;
     }
@@ -5149,11 +5150,11 @@ fn vertical_align(
         VerticalAlign::Bottom => rest,
         VerticalAlign::Center => rest / 2,
     };
-    let mut output = Vec::with_capacity(height);
+    LinePlan::from_lines((0..top).map(|_| Line::blank(width)))
-    output.extend((0..top).map(|_| Line::blank(width)));
+        .concat(&lines)
-    output.append(&mut lines);
+        .concat(&LinePlan::from_lines(
-    output.extend((0..(rest - top)).map(|_| Line::blank(width)));
+            (0..(rest - top)).map(|_| Line::blank(width)),
-    output
+        ))
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -5824,7 +5825,7 @@ fn flex_resolved_align(item: &FlexRuntimeItem<'_>, container: FlexAlign) -> Flex
 }
 
 fn pad_rendered_width(mut rendered: Rendered, width: i64, align: FlexAlign) -> Rendered {
-    for line in &mut rendered.lines {
+    rendered.lines = rendered.lines.map_lines(|_, mut line| {
         let extra = (width - line.width).max(0);
         let left = match align {
             FlexAlign::FlexEnd | FlexAlign::End | FlexAlign::Right | FlexAlign::SelfEnd => extra,
@@ -5833,7 +5834,8 @@ fn pad_rendered_width(mut rendered: Rendered, width: i64, align: FlexAlign) -> R
         };
         line.prepend_space(left);
         line.push_space(extra - left);
-    }
+        line
+    });
     rendered
 }
 
@@ -5960,38 +5962,40 @@ fn concat_horizontal_sized(parts: Vec<(Rendered, i64)>, target_height: i64) -> R
         .max()
         .unwrap_or(1)
         .max(target_height);
-    let mut lines = Vec::with_capacity(height as usize);
+    let mut parts = parts
-    for index in 0..height as usize {
+        .iter()
+        .map(|(rendered, width)| (rendered.lines.iter(), *width))
+        .collect::<Vec<_>>();
+    let lines = LinePlan::from_lines((0..height as usize).map(|_| {
         let mut line = Line::default();
-        for (rendered, width) in &parts {
+        for (lines, width) in &mut parts {
-            if let Some(part) = rendered.lines.get(index) {
+            if let Some(part) = lines.next() {
                 line.append(part);
             } else {
                 line.push_space(*width);
             }
         }
-        lines.push(line);
+        line
-    }
+    }));
-    Rendered::from_lines(lines)
+    Rendered { lines }
 }
 
 fn stack_vertical(parts: Vec<Rendered>) -> Rendered {
-    let mut lines = Vec::new();
+    fn join_parts(parts: &[Rendered]) -> LinePlan {
-    let mut breaks = Vec::new();
+        match parts {
-    for mut part in parts {
+            [] => LinePlan::default(),
-        if part.lines.is_empty() {
+            [part] => part.lines.clone(),
-            continue;
+            _ => {
+                let middle = parts.len() / 2;
+                join_parts(&parts[..middle]).concat(&join_parts(&parts[middle..]))
+            }
         }
-        if !lines.is_empty() {
-            breaks.push(AtomProperties::default());
-        }
-        lines.append(&mut part.lines);
-        breaks.append(&mut part.breaks);
     }
+    let mut lines = join_parts(&parts);
     if lines.is_empty() {
-        lines.push(Line::default());
+        lines = LinePlan::from_lines([Line::default()]);
     }
-    Rendered { lines, breaks }
+    Rendered { lines }
 }
 
 fn slice_rendered(rendered: Rendered, start: i64, height: i64) -> Rendered {
@@ -5999,25 +6003,9 @@ fn slice_rendered(rendered: Rendered, start: i64, height: i64) -> Rendered {
     let height = usize::try_from(height.max(0)).unwrap_or(0);
     let end = start.saturating_add(height).min(rendered.lines.len());
     if start >= end {
-        return Rendered {
+        return Rendered::from_line_plan(LinePlan::default());
-            lines: Vec::new(),
-            breaks: Vec::new(),
-        };
     }
-    let line_count = end - start;
+    Rendered::from_line_plan(rendered.lines.slice(start..end))
-    let lines = rendered
-        .lines
-        .into_iter()
-        .skip(start)
-        .take(line_count)
-        .collect::<Vec<_>>();
-    let breaks = rendered
-        .breaks
-        .into_iter()
-        .skip(start)
-        .take(line_count.saturating_sub(1))
-        .collect::<Vec<_>>();
-    Rendered { lines, breaks }
 }
 
 fn exact_rendered_height(
@@ -6113,14 +6101,12 @@ fn render_column_window(
     start: i64,
     height: i64,
 ) -> Result<Rendered, String> {
-    let mut leaves = Vec::new();
-    collect_column_leaves(children, &mut leaves);
     let end = start.saturating_add(height).max(start);
     let target = context.viewport_width.max(0);
     let mut offset = 0_i64;
     let mut parts = Vec::new();
 
-    for child in leaves {
+    for child in column_leaves(children) {
         let child_height = exact_rendered_height(child, resolver, context)
             .ok_or_else(|| "Native layout column window has an unbounded child".to_owned())?;
         let child_end = offset.saturating_add(child_height);
@@ -6152,12 +6138,14 @@ fn render_column_window(
         };
 
         let extra = (target - rendered.first_width()).max(0);
-        for line in &mut rendered.lines {
-            line.push_space(extra);
-        }
         if extra > 0 {
-            rendered.breaks =
+            rendered.lines = rendered
-                vec![AtomProperties::default(); rendered.lines.len().saturating_sub(1)];
+                .lines
+                .map_lines(|_, mut line| {
+                    line.push_space(extra);
+                    line
+                })
+                .clear_breaks();
         }
         if !rendered.lines.is_empty() {
             parts.push(rendered);
@@ -6856,10 +6844,11 @@ fn render_node_with_override(
                 });
             if transparent_preformatted {
                 let mut rendered = child_rendered.take().expect("validated child");
-                for (index, line) in rendered.lines.iter_mut().enumerate() {
+                rendered.lines = rendered.lines.map_lines(|index, mut line| {
                     line.own_content(*region_id, index as i64);
                     line.apply_property_template(*surface_template_id);
-                }
+                    line
+                });
                 return Ok(rendered);
             }
 
@@ -6873,10 +6862,8 @@ fn render_node_with_override(
                         line.apply_property_template(*content_surface_template_id);
                     }
                 }
-                content_lines
+                LinePlan::from_lines(content_lines)
-                    .into_iter()
+                    .map_lines(|_, line| line.padded(content_width, *text_align))
-                    .map(|line| line.padded(content_width, *text_align))
-                    .collect::<Vec<_>>()
             } else {
                 let rendered = child_rendered.expect("validated child");
                 let uniform_width = rendered
@@ -6889,18 +6876,17 @@ fn render_node_with_override(
                 } else {
                     rendered
                 };
-                rendered
+                if preserve_exact_width {
-                    .lines
+                    rendered.lines
-                    .into_iter()
+                } else {
-                    .map(|line| {
+                    rendered
-                        if preserve_exact_width {
+                        .lines
-                            line
+                        .map_lines(|_, line| line.padded(content_width, *text_align))
-                        } else {
+                }
-                            line.padded(content_width, *text_align)
-                        }
-                    })
-                    .collect::<Vec<_>>()
             };
+            // Ordinary Box formatting historically rebuilds default breaks;
+            // the transparent preformatted branch above preserves child breaks.
+            formatted = formatted.clear_breaks();
             let text_height = windowed_child_height.unwrap_or(formatted.len() as i64);
             let minimum_height = resolve_height(
                 min_height,
@@ -6953,7 +6939,7 @@ fn render_node_with_override(
                 && *vertical == VerticalAlign::Top;
 
             let mut content_index_start = 0_i64;
-            let mut overflow_lines = Vec::new();
+            let mut overflow_lines = LinePlan::default();
             if let Some(start) = windowed_child_start {
                 content_index_start = start;
             } else if formatted.len() > content_height as usize {
@@ -6964,48 +6950,46 @@ fn render_node_with_override(
                         if simple_scroll_rendered {
                             content_index_start = start as i64;
                         }
-                        formatted = formatted[start..start + content_height as usize].to_vec();
+                        formatted = formatted.slice(start..start + content_height as usize);
                     }
                     Overflow::Hidden => {
-                        formatted.truncate(content_height as usize);
+                        formatted = formatted.slice(0..content_height as usize);
                     }
                     Overflow::Visible => {
-                        overflow_lines = formatted.split_off(content_height as usize);
+                        overflow_lines = formatted.slice(content_height as usize..formatted.len());
+                        formatted = formatted.slice(0..content_height as usize);
                     }
                 }
             }
-            for (index, line) in formatted.iter_mut().enumerate() {
+            formatted = formatted.map_lines(|index, mut line| {
                 line.own_content(*region_id, content_index_start + index as i64);
-            }
+                line
-            let mut lines =
+            });
+            let lines =
                 vertical_align(formatted, content_height as usize, *vertical, content_width)
-                    .into_iter()
+                    .map_lines(|_, line| {
-                    .map(|line| {
                         if simple_scroll_rendered {
                             line
                         } else {
                             line.collapse_whitespace_content(content_width, *region_id)
                         }
-                    })
+                    });
-                    .collect::<Vec<_>>();
 
-            let mut padded =
+            let mut padded = LinePlan::from_lines((0..*padding_top).map(|_| {
-                Vec::with_capacity(lines.len() + *padding_top as usize + *padding_bottom as usize);
-            padded.extend((0..*padding_top).map(|_| {
                 Line::blank_with_properties(
                     content_width,
                     region_properties(RegionRole::PaddingTop, *region_id, None),
                 )
             }));
-            padded.append(&mut lines);
+            padded = padded.concat(&lines);
-            padded.extend((0..*padding_bottom).map(|_| {
+            padded = padded.concat(&LinePlan::from_lines((0..*padding_bottom).map(|_| {
                 Line::blank_with_properties(
                     content_width,
                     region_properties(RegionRole::PaddingBottom, *region_id, None),
                 )
-            }));
+            })));
 
-            for line in &mut padded {
+            padded = padded.map_lines(|_, mut line| {
                 line.prepend_space_with_properties(
                     *padding_left,
                     region_properties(RegionRole::PaddingLeft, *region_id, None),
@@ -7025,23 +7009,27 @@ fn render_node_with_override(
                     *border_right,
                     region_properties(RegionRole::BorderRight, *region_id, *border_right_style),
                 );
+                line
+            });
+            if border_top_style.is_some() && !padded.is_empty() {
+                padded = padded.update_line(0, |mut line| {
+                    line.apply_style(*border_top_style);
+                    line.apply_role(RegionRole::BorderTop, *region_id);
+                    line
+                });
             }
-            if border_top_style.is_some() {
+            if border_bottom_style.is_some() && !padded.is_empty() {
-                if let Some(first) = padded.first_mut() {
+                padded = padded.update_line(padded.len() - 1, |mut line| {
-                    first.apply_style(*border_top_style);
+                    line.apply_style(*border_bottom_style);
-                    first.apply_role(RegionRole::BorderTop, *region_id);
+                    line.apply_role(RegionRole::BorderBottom, *region_id);
-                }
+                    line
+                });
             }
-            if border_bottom_style.is_some() {
+            padded = padded.map_lines(|_, mut line| {
-                if let Some(last) = padded.last_mut() {
-                    last.apply_style(*border_bottom_style);
-                    last.apply_role(RegionRole::BorderBottom, *region_id);
-                }
-            }
-            for line in &mut padded {
                 line.apply_property_template(*surface_template_id);
-            }
+                line
-            for line in &mut padded {
+            });
+            padded = padded.map_lines(|_, mut line| {
                 line.prepend_space_with_properties(
                     *margin_left,
                     region_properties(RegionRole::MarginLeft, *region_id, None),
@@ -7050,33 +7038,33 @@ fn render_node_with_override(
                     *margin_right,
                     region_properties(RegionRole::MarginRight, *region_id, None),
                 );
-            }
+                line
+            });
             let total_width = content_width + side_width;
-            let mut output =
+            let mut output = LinePlan::from_lines((0..*margin_top).map(|_| {
-                Vec::with_capacity(padded.len() + *margin_top as usize + *margin_bottom as usize);
-            output.extend((0..*margin_top).map(|_| {
                 Line::blank_with_properties(
                     total_width,
                     region_properties(RegionRole::MarginTop, *region_id, None),
                 )
             }));
-            output.append(&mut padded);
+            output = output.concat(&padded);
-            output.extend((0..*margin_bottom).map(|_| {
+            output = output.concat(&LinePlan::from_lines((0..*margin_bottom).map(|_| {
                 Line::blank_with_properties(
                     total_width,
                     region_properties(RegionRole::MarginBottom, *region_id, None),
                 )
-            }));
+            })));
-            let mut rendered = Rendered::from_lines(output);
+            let mut rendered = Rendered::from_line_plan(output);
             if !overflow_lines.is_empty() {
                 let left_space = margin_left + border_left + padding_left;
                 let right_space = padding_right + border_right + margin_right;
-                for line in &mut overflow_lines {
+                overflow_lines = overflow_lines.map_lines(|_, mut line| {
                     line.apply_style(*foreground_style);
                     line.prepend_space(left_space);
                     line.push_space(right_space);
-                }
+                    line
-                rendered = stack_vertical(vec![rendered, Rendered::from_lines(overflow_lines)]);
+                });
+                rendered = stack_vertical(vec![rendered, Rendered::from_line_plan(overflow_lines)]);
             }
             if *overflow == Overflow::Scroll && text_height > content_height {
                 rendered.apply_scroll_window(*region_id);
@@ -7111,30 +7099,17 @@ fn render_node_with_override(
                 .iter()
                 .map(|child| render_node(child, resolver, child_context, intrinsic))
                 .collect::<Result<Vec<_>, _>>()?;
-            let height = rendered
+            let parts = rendered
-                .iter()
+                .into_iter()
-                .map(|item| item.lines.len())
+                .map(|rendered| {
-                .max()
+                    let width = rendered.first_width();
-                .unwrap_or(1);
+                    (rendered, width)
-            let mut lines = Vec::with_capacity(height);
+                })
-            for index in 0..height {
+                .collect();
-                let mut line = Line::default();
+            Ok(concat_horizontal_sized(parts, 0))
-                for item in &rendered {
-                    if let Some(child_line) = item.lines.get(index) {
-                        line.append(child_line);
-                    } else {
-                        line.push_space(item.first_width());
-                    }
-                }
-                lines.push(line);
-            }
-            Ok(Rendered::from_lines(lines))
         }
         LayoutNode::Column { children, .. } => {
-            let mut leaves = Vec::new();
+            let rendered = column_leaves(children)
-            collect_column_leaves(children, &mut leaves);
-            let rendered = leaves
-                .into_iter()
                 .map(|child| render_node(child, resolver, context, intrinsic))
                 .collect::<Result<Vec<_>, _>>()?;
             let maximum = rendered
@@ -7153,12 +7128,14 @@ fn render_node_with_override(
             let mut parts = Vec::with_capacity(rendered.len());
             for mut item in rendered {
                 let extra = (target - item.first_width()).max(0);
-                for line in &mut item.lines {
-                    line.push_space(extra);
-                }
                 if extra > 0 {
-                    item.breaks =
+                    item.lines = item
-                        vec![AtomProperties::default(); item.lines.len().saturating_sub(1)];
+                        .lines
+                        .map_lines(|_, mut line| {
+                            line.push_space(extra);
+                            line
+                        })
+                        .clear_breaks();
                 }
                 parts.push(item);
             }
@@ -7194,14 +7171,19 @@ fn render_node_with_override(
     }
 }
 
-fn collect_column_leaves<'a>(children: &'a [LayoutNode], output: &mut Vec<&'a LayoutNode>) {
+// Flatten only literal Columns. In particular, a retained NodeRef remains a
-    for child in children {
+// child evaluation boundary even when it resolves to an identified Column.
-        if let LayoutNode::Column { children, .. } = child {
+fn column_leaves(children: &[LayoutNode]) -> impl Iterator<Item = &LayoutNode> {
-            collect_column_leaves(children, output);
+    let mut stack = vec![children.iter()];
-        } else {
+    std::iter::from_fn(move || loop {
-            output.push(child);
+        match stack.last_mut()?.next() {
+            Some(LayoutNode::Column { children, .. }) => stack.push(children.iter()),
+            Some(child) => return Some(child),
+            None => {
+                stack.pop();
+            }
         }
-    }
+    })
 }
 
 #[cfg(test)]
@@ -7711,14 +7693,14 @@ mod tests {
             }),
         )
         .expect("typed column render");
-        assert_eq!(rendered.lines[0].width, 100);
+        assert_eq!(rendered.lines.get(0).unwrap().width, 100);
         assert!(
             matches!(
-                rendered.lines[0].atoms.first(),
+                rendered.lines.get(0).unwrap().atoms.first(),
                 Some(Atom::Space { width: 45, .. })
             ),
             "{:?}",
-            rendered.lines[0].atoms
+            rendered.lines.get(0).unwrap().atoms
         );
     }
 
@@ -8210,18 +8192,28 @@ mod tests {
     #[test]
     fn vertical_stack_and_scroll_preserve_break_metadata() {
         let mut first = Rendered::from_lines(vec![Line::blank(1), Line::blank(1)]);
-        first.breaks[0].owner = Some(11);
+        first.lines = first.lines.with_break(
+            0,
+            AtomProperties {
+                owner: Some(11),
+                ..AtomProperties::default()
+            },
+        );
         let second = Rendered::from_lines(vec![Line::blank(1)]);
         let mut stacked = stack_vertical(vec![first, second]);
 
-        assert_eq!(stacked.breaks.len(), 2);
+        assert_eq!(stacked.lines.len() - 1, 2);
-        assert_eq!(stacked.breaks[0].owner, Some(11));
+        assert_eq!(stacked.lines.break_after(0).unwrap().owner, Some(11));
-        assert_eq!(stacked.breaks[1], AtomProperties::default());
+        assert_eq!(
+            stacked.lines.break_after(1),
+            Some(&AtomProperties::default())
+        );
 
         stacked.apply_scroll_window(9);
         assert!(stacked
-            .breaks
+            .lines
-            .iter()
+            .iter_with_breaks()
+            .filter_map(|(_, properties)| properties)
             .all(|properties| { properties.scroll_window == Some(9) }));
         assert!(stacked.lines.iter().all(|line| {
             line.atoms
@@ -8239,6 +8231,233 @@ mod tests {
         assert!(tape.lines[2].break_after.is_none());
     }
 
+    #[test]
+    fn line_sequence_oracle_wrap_slice_and_join_keep_distinct_breaks() {
+        let source = Rendered {
+            lines: LinePlan::from_lines([
+                Line::from_clusters(&[cluster("a", 2, None), cluster("b", 2, None)]),
+                Line::from_clusters(&[cluster("c", 2, None), cluster("d", 2, None)]),
+            ])
+            .with_break(
+                0,
+                AtomProperties {
+                    owner: Some(17),
+                    scroll_window: Some(9),
+                    ..AtomProperties::default()
+                },
+            ),
+        };
+        let wrapped = wrap_rendered(source, 2, WrapMode::Char);
+        let tape = wrapped.clone().into_tape(0);
+        assert_eq!(tape.lines.len(), 4);
+        assert_eq!(tape.lines[0].break_after, Some(AtomProperties::default()));
+        assert_eq!(tape.lines[1].break_after.as_ref().unwrap().owner, Some(17));
+        assert_eq!(tape.lines[2].break_after, Some(AtomProperties::default()));
+        let middle = slice_rendered(wrapped, 1, 2);
+        assert_eq!(
+            middle.clone().into_tape(0).lines,
+            tape.lines[1..3]
+                .iter()
+                .cloned()
+                .enumerate()
+                .map(|(index, mut line)| {
+                    if index == 1 {
+                        line.break_after = None;
+                    }
+                    line
+                })
+                .collect::<Vec<_>>()
+        );
+        let joined =
+            stack_vertical(vec![middle, Rendered::from_lines(vec![Line::blank(1)])]).into_tape(0);
+        assert_eq!(
+            joined.lines[0].break_after.as_ref().unwrap().owner,
+            Some(17)
+        );
+        assert_eq!(joined.lines[1].break_after, Some(AtomProperties::default()));
+    }
+
+    fn nonuniform_text(region_id: i64, widths: &[i64]) -> LayoutNode {
+        LayoutNode::Text {
+            node_id: None,
+            node_revision: None,
+            region_id,
+            content: Arc::new(measured_text(
+                widths
+                    .iter()
+                    .map(|width| vec![cluster("x", *width, None)])
+                    .collect(),
+            )),
+            typography_style: None,
+            foreground_style: None,
+            surface_template_id: None,
+            wrap_mode: WrapMode::None,
+        }
+    }
+
+    #[test]
+    fn line_sequence_oracle_column_uses_first_width_and_constant_child_extra() {
+        let column = LayoutNode::Column {
+            node_id: None,
+            node_revision: None,
+            children: Arc::new(vec![
+                nonuniform_text(1, &[2, 9]),
+                nonuniform_text(2, &[5, 1]),
+            ]),
+        };
+        let tape = render_node(&column, None, test_context(), true)
+            .unwrap()
+            .into_tape(0);
+        assert_eq!(
+            tape.lines.iter().map(|line| line.width).collect::<Vec<_>>(),
+            [5, 12, 5, 1]
+        );
+        let tape = render_node(
+            &column,
+            None,
+            LayoutContext {
+                viewport_width: 3,
+                ..test_context()
+            },
+            false,
+        )
+        .unwrap()
+        .into_tape(0);
+        assert_eq!(
+            tape.lines.iter().map(|line| line.width).collect::<Vec<_>>(),
+            [3, 10, 5, 1]
+        );
+    }
+
+    #[test]
+    fn line_sequence_oracle_column_flattens_literal_columns_but_not_node_refs() {
+        let inner = identified(
+            LayoutNode::Column {
+                node_id: None,
+                node_revision: None,
+                children: Arc::new(vec![nonuniform_text(1, &[2]), nonuniform_text(2, &[5])]),
+            },
+            2,
+            1,
+        );
+        let root = identified(
+            LayoutNode::Column {
+                node_id: None,
+                node_revision: None,
+                children: Arc::new(vec![inner, nonuniform_text(3, &[8])]),
+            },
+            1,
+            1,
+        );
+        let document = retained_document(root);
+        let (retained, _) = RetainedDocument::bootstrap(document.clone()).unwrap();
+        let full = render_node(&document.root, None, test_context(), true)
+            .unwrap()
+            .into_tape(0);
+        let referenced = render_node(
+            &LayoutNode::NodeRef {
+                node_id: retained.root_id,
+            },
+            Some(&retained),
+            test_context(),
+            true,
+        )
+        .unwrap()
+        .into_tape(0);
+        assert_eq!(
+            full.lines
+                .iter()
+                .map(|line| line.atoms.len())
+                .collect::<Vec<_>>(),
+            [2, 2, 1]
+        );
+        assert_eq!(
+            referenced
+                .lines
+                .iter()
+                .map(|line| line.atoms.len())
+                .collect::<Vec<_>>(),
+            [3, 2, 1]
+        );
+        assert_eq!(
+            full.lines.iter().map(|line| line.width).collect::<Vec<_>>(),
+            [8, 8, 8]
+        );
+        assert_eq!(
+            referenced
+                .lines
+                .iter()
+                .map(|line| line.width)
+                .collect::<Vec<_>>(),
+            [8, 8, 8]
+        );
+    }
+
+    #[test]
+    fn line_sequence_oracle_column_only_positive_extra_clears_scroll_breaks() {
+        let mut scrolled = child_box(11, nonuniform_text(12, &[2, 2, 2]), None);
+        let LayoutNode::Box { height, .. } = &mut scrolled else {
+            unreachable!()
+        };
+        *height = Size::Lines { value: 2 };
+        let column = LayoutNode::Column {
+            node_id: None,
+            node_revision: None,
+            children: Arc::new(vec![scrolled]),
+        };
+        for (width, expected_scroll) in [(1, Some(11)), (2, Some(11)), (3, None)] {
+            let tape = render_node(
+                &column,
+                None,
+                LayoutContext {
+                    viewport_width: width,
+                    ..test_context()
+                },
+                false,
+            )
+            .unwrap()
+            .into_tape(0);
+            assert_eq!(
+                tape.lines[0].break_after.as_ref().unwrap().scroll_window,
+                expected_scroll
+            );
+        }
+    }
+
+    #[test]
+    fn line_sequence_oracle_horizontal_missing_lines_use_first_width() {
+        let row = LayoutNode::Row {
+            node_id: None,
+            node_revision: None,
+            children: Arc::new(vec![
+                nonuniform_text(1, &[2, 9]),
+                nonuniform_text(2, &[5, 1, 3]),
+            ]),
+        };
+        let tape = render_node(&row, None, test_context(), true)
+            .unwrap()
+            .into_tape(0);
+        assert_eq!(
+            tape.lines.iter().map(|line| line.width).collect::<Vec<_>>(),
+            [7, 10, 5]
+        );
+        let parts = vec![
+            (
+                Rendered::from_lines(vec![Line::blank(2), Line::blank(9)]),
+                4,
+            ),
+            (
+                Rendered::from_lines(vec![Line::blank(5), Line::blank(1), Line::blank(3)]),
+                6,
+            ),
+        ];
+        let tape = concat_horizontal_sized(parts, 4).into_tape(0);
+        assert_eq!(
+            tape.lines.iter().map(|line| line.width).collect::<Vec<_>>(),
+            [7, 10, 7, 10]
+        );
+    }
+
     #[test]
     fn binary_tapes_are_versioned_bounded_and_support_light_frames() {
         let properties = AtomProperties {
@@ -8258,7 +8477,7 @@ mod tests {
             Line::blank_with_properties(8, properties.clone()),
             Line::blank_with_properties(8, properties.clone()),
         ]);
-        rendered.breaks[0] = properties;
+        rendered.lines = rendered.lines.with_break(0, properties);
         let tape = rendered.into_tape(3);
         let styles = vec![
             StyleTemplate {

native/src/lib.rs

@@ -470,6 +470,7 @@ struct RenderedJob {
     target_renders: u64,
     resolver_lookups: u64,
     atom_plan_work: layout::AtomPlanWork,
+    line_plan_work: layout::LinePlanWork,
 }
 
 #[derive(Default)]
@@ -523,6 +524,7 @@ struct RuntimeState {
     document_trie_path_nodes_copied: u64,
     document_resolver_lookups: u64,
     atom_plan_work: layout::AtomPlanWork,
+    line_plan_work: layout::LinePlanWork,
     confirmed_baseline: Option<Arc<ConfirmedBaseline>>,
 }
 
@@ -547,6 +549,23 @@ struct Shared {
     result_available: Condvar,
 }
 
+impl Shared {
+    fn request_stop(&self) {
+        // Both worker condition variables inspect `alive` under this mutex.
+        // Publish shutdown under the same lock so notification cannot fall
+        // between a worker's predicate check and its atomic unlock-and-wait.
+        {
+            let _state = self
+                .state
+                .lock()
+                .unwrap_or_else(|poison| poison.into_inner());
+            self.alive.store(false, Ordering::Release);
+        }
+        self.job_available.notify_all();
+        self.result_available.notify_all();
+    }
+}
+
 #[derive(Debug)]
 struct Session {
     shared: Arc<Shared>,
@@ -584,6 +603,7 @@ struct SessionStats {
     document_trie_path_nodes_copied: u64,
     document_resolver_lookups: u64,
     atom_plan_work: layout::AtomPlanWork,
+    line_plan_work: layout::LinePlanWork,
     pending_baselines: usize,
     confirmed_baseline: bool,
     confirmed_baseline_bytes: usize,
@@ -657,6 +677,7 @@ impl Session {
                 document_trie_path_nodes_copied: 0,
                 document_resolver_lookups: 0,
                 atom_plan_work: layout::AtomPlanWork::default(),
+                line_plan_work: layout::LinePlanWork::default(),
                 confirmed_baseline,
             }),
             readiness_channel: Mutex::new(None),
@@ -672,8 +693,7 @@ impl Session {
             match spawn {
                 Ok(handle) => handles.push(handle),
                 Err(error) => {
-                    shared.alive.store(false, Ordering::Release);
+                    shared.request_stop();
-                    shared.job_available.notify_all();
                     for handle in handles {
                         let _ = handle.join();
                     }
@@ -1083,6 +1103,7 @@ impl Session {
         state.target_renders += output.target_renders;
         state.document_resolver_lookups += output.resolver_lookups;
         state.atom_plan_work.accumulate(output.atom_plan_work);
+        state.line_plan_work.accumulate(output.line_plan_work);
         state.document_parses += input_stats.parses;
         state.document_validations += input_stats.validations;
         state.document_reuses += input_stats.reuses;
@@ -1232,6 +1253,7 @@ impl Session {
             document_trie_path_nodes_copied: state.document_trie_path_nodes_copied,
             document_resolver_lookups: state.document_resolver_lookups,
             atom_plan_work: state.atom_plan_work,
+            line_plan_work: state.line_plan_work,
             pending_baselines,
             confirmed_baseline: state.confirmed_baseline.is_some(),
             confirmed_baseline_bytes,
@@ -1246,9 +1268,7 @@ impl Session {
 
     fn stop(&self, join: bool) {
         self.detach_readiness_channel();
-        self.shared.alive.store(false, Ordering::Release);
+        self.shared.request_stop();
-        self.shared.job_available.notify_all();
-        self.shared.result_available.notify_all();
         if let Some(handles) = self
             .workers
             .lock()
@@ -1445,6 +1465,7 @@ fn render_layout_payload(
             target_renders: 0,
             resolver_lookups: 0,
             atom_plan_work: layout::AtomPlanWork::default(),
+            line_plan_work: layout::LinePlanWork::default(),
         },
         JobPayload::Layout {
             document,
@@ -1503,6 +1524,7 @@ fn render_layout_payload(
             >;
             layout::reset_resolver_lookups();
             layout::reset_atom_plan_work();
+            layout::reset_line_plan_work();
             let result = catch_unwind(AssertUnwindSafe(|| -> LayoutRenderOutcome {
                 let target_styles = document.styles()?;
                 if let Some(base_context) = base_context {
@@ -1578,6 +1600,7 @@ fn render_layout_payload(
             let resolver_lookups =
                 validation_resolver_lookups.saturating_add(layout::resolver_lookups());
             let atom_plan_work = layout::atom_plan_work();
+            let line_plan_work = layout::line_plan_work();
             match result {
                 Ok(Ok((bytes, tape, styles, baseline_hit, base_renders, target_renders))) => {
                     RenderedJob {
@@ -1594,6 +1617,7 @@ fn render_layout_payload(
                         target_renders,
                         resolver_lookups,
                         atom_plan_work,
+                        line_plan_work,
                     }
                 }
                 Ok(Err(error)) => RenderedJob {
@@ -1604,6 +1628,7 @@ fn render_layout_payload(
                     target_renders: 0,
                     resolver_lookups,
                     atom_plan_work,
+                    line_plan_work,
                 },
                 Err(_) => RenderedJob {
                     bytes: encode_error_tape(identity, "native layout panicked", max_result_bytes),
@@ -1613,6 +1638,7 @@ fn render_layout_payload(
                     target_renders: 0,
                     resolver_lookups,
                     atom_plan_work,
+                    line_plan_work,
                 },
             }
         }
@@ -1763,6 +1789,7 @@ fn worker_loop(shared: Arc<Shared>) {
         state.target_renders += output.target_renders;
         state.document_resolver_lookups += output.resolver_lookups;
         state.atom_plan_work.accumulate(output.atom_plan_work);
+        state.line_plan_work.accumulate(output.line_plan_work);
         state.results.insert(
             (job.generation, job.key),
             ResultEntry {
@@ -2358,6 +2385,69 @@ mod tests {
         session.stop(true);
     }
 
+    #[test]
+    fn stopping_serializes_with_a_worker_holding_the_wait_predicate_lock() {
+        let session = Arc::new(Session::new(1, 1, 1, 4).unwrap());
+        let guard = session.shared.state.lock().unwrap();
+        let (started_tx, started_rx) = std::sync::mpsc::channel();
+        let (completed_tx, completed_rx) = std::sync::mpsc::channel();
+        let stopping = Arc::clone(&session);
+        let stopper = thread::spawn(move || {
+            started_tx.send(()).unwrap();
+            stopping.stop(false);
+            completed_tx.send(()).unwrap();
+        });
+        started_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        // A worker holds this mutex between checking the predicate and
+        // entering Condvar::wait. Shutdown must not publish/notify in that gap.
+        let completed_while_locked = completed_rx
+            .recv_timeout(Duration::from_millis(100))
+            .is_ok();
+        drop(guard);
+        if !completed_while_locked {
+            completed_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        }
+        stopper.join().unwrap();
+        assert!(
+            !completed_while_locked,
+            "shutdown raced ahead of the worker wait predicate lock"
+        );
+    }
+
+    #[test]
+    fn stopping_joins_a_worker_blocked_on_result_capacity() {
+        let session = Arc::new(Session::new(1, 2, 1, 8).unwrap());
+        session
+            .submit(
+                1,
+                &batch(r#"[{"key":1,"payload":"a"},{"key":2,"payload":"b"}]"#),
+            )
+            .unwrap();
+        wait_until(|| {
+            let stats = session.stats();
+            stats.completed_jobs == 1 && stats.queued_jobs == 0
+        });
+        let stopping = Arc::clone(&session);
+        let (completed_tx, completed_rx) = std::sync::mpsc::channel();
+        let stopper = thread::spawn(move || {
+            stopping.stop(true);
+            completed_tx.send(()).unwrap();
+        });
+        let completed_without_rescue = completed_rx.recv_timeout(Duration::from_secs(2)).is_ok();
+        if !completed_without_rescue {
+            // Preserve the failure, but release a stranded waiter before the
+            // assertion so a broken notification does not hang the test suite.
+            session.shared.job_available.notify_all();
+            session.shared.result_available.notify_all();
+            completed_rx.recv_timeout(Duration::from_secs(2)).unwrap();
+        }
+        stopper.join().unwrap();
+        assert!(
+            completed_without_rescue,
+            "shutdown left a capacity waiter asleep"
+        );
+    }
+
     #[test]
     fn malformed_and_over_capacity_batches_are_rejected() {
         let session = Session::new(1, 1, 1, 4).unwrap();

native/src/line_plan.rs

@@ -0,0 +1,599 @@
+use std::cell::Cell;
+use std::ops::Range;
+use std::sync::Arc;
+
+use crate::sequence::{Entry, Iter, Measure, Sequence, Work};
+
+use super::{AtomProperties, Line};
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) struct LinePlanWork {
+    pub(crate) sequence_nodes_created: u64,
+    pub(crate) sequence_entries_written: u64,
+    pub(crate) iterator_nodes_visited: u64,
+    pub(crate) lines_visited: u64,
+    pub(crate) line_lookups: u64,
+    pub(crate) range_queries: u64,
+    pub(crate) lines_mapped: u64,
+    pub(crate) breaks_mapped: u64,
+    pub(crate) line_clones: u64,
+    pub(crate) materialized_lines: u64,
+    pub(crate) materialized_breaks: u64,
+    pub(crate) materialized_chars: u64,
+}
+
+thread_local! {
+    static WORK: Cell<LinePlanWork> = Cell::new(LinePlanWork::default());
+}
+
+pub(crate) fn reset_work() {
+    WORK.set(LinePlanWork::default());
+}
+
+pub(crate) fn work() -> LinePlanWork {
+    WORK.get()
+}
+
+impl LinePlanWork {
+    pub(crate) fn accumulate(&mut self, other: Self) {
+        macro_rules! add {
+            ($($field:ident),+ $(,)?) => {
+                $(self.$field = self.$field.checked_add(other.$field)
+                    .expect("native line plan work counter overflow");)+
+            };
+        }
+        add!(
+            sequence_nodes_created,
+            sequence_entries_written,
+            iterator_nodes_visited,
+            lines_visited,
+            line_lookups,
+            range_queries,
+            lines_mapped,
+            breaks_mapped,
+            line_clones,
+            materialized_lines,
+            materialized_breaks,
+            materialized_chars
+        );
+    }
+}
+
+fn record(delta: LinePlanWork) {
+    let mut work = WORK.get();
+    work.accumulate(delta);
+    WORK.set(work);
+}
+
+fn record_sequence(work: Work) {
+    record(LinePlanWork {
+        sequence_nodes_created: work.nodes_created,
+        sequence_entries_written: work.entries_written,
+        ..LinePlanWork::default()
+    });
+}
+
+pub(super) fn record_materialized(lines: usize, chars: u64) {
+    record(LinePlanWork {
+        materialized_lines: lines as u64,
+        materialized_breaks: lines.saturating_sub(1) as u64,
+        materialized_chars: chars,
+        ..LinePlanWork::default()
+    });
+}
+
+pub(super) fn clone_line(line: &Line) -> Line {
+    record(LinePlanWork {
+        line_clones: 1,
+        ..LinePlanWork::default()
+    });
+    line.clone()
+}
+
+#[derive(Debug)]
+struct LineEntry {
+    line: Line,
+    // Stored even at the terminal entry, where it is not part of the output.
+    // Concatenation replaces that boundary explicitly before it becomes visible.
+    break_after: Arc<AtomProperties>,
+}
+
+fn entry(line: Line, break_after: Arc<AtomProperties>) -> Entry<LineEntry> {
+    // One virtual newline per entry makes prefixes additive. The final newline
+    // is removed by each output-range summary, including empty/zero-width lines.
+    let chars = line
+        .atoms
+        .char_count()
+        .checked_add(1)
+        .expect("validated native line character count fits u64");
+    let measure = Measure::new(chars, 1, line.width, line.width)
+        .expect("validated native line width is nonnegative");
+    Entry::new(Arc::new(LineEntry { line, break_after }), measure)
+        .expect("validated native line measure")
+}
+
+#[derive(Clone, Debug, Default)]
+pub(super) struct LinePlan {
+    entries: Sequence<LineEntry>,
+}
+
+/// Borrowed traversal; counts both sequence-node visits and yielded lines.
+pub(super) struct Lines<'a> {
+    entries: Iter<'a, LineEntry>,
+}
+
+impl<'a> Iterator for Lines<'a> {
+    type Item = &'a Line;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let before = self.entries.nodes_visited();
+        let value = self.entries.next();
+        record(LinePlanWork {
+            iterator_nodes_visited: self.entries.nodes_visited() - before,
+            lines_visited: u64::from(value.is_some()),
+            ..LinePlanWork::default()
+        });
+        value.map(|entry| &entry.value().line)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.entries.size_hint()
+    }
+}
+
+impl ExactSizeIterator for Lines<'_> {}
+
+pub(super) struct LinesAndBreaks<'a> {
+    lines: Lines<'a>,
+}
+
+impl<'a> Iterator for LinesAndBreaks<'a> {
+    type Item = (&'a Line, Option<&'a AtomProperties>);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let before = self.lines.entries.nodes_visited();
+        let value = self.lines.entries.next();
+        record(LinePlanWork {
+            iterator_nodes_visited: self.lines.entries.nodes_visited() - before,
+            lines_visited: u64::from(value.is_some()),
+            ..LinePlanWork::default()
+        });
+        let entry = value?.value();
+        Some((
+            &entry.line,
+            (self.lines.entries.len() > 0).then_some(entry.break_after.as_ref()),
+        ))
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.lines.size_hint()
+    }
+}
+
+impl ExactSizeIterator for LinesAndBreaks<'_> {}
+
+impl LinePlan {
+    pub(super) fn from_lines(lines: impl IntoIterator<Item = Line>) -> Self {
+        let default_break = Arc::new(AtomProperties::default());
+        Self::from_entries(
+            lines
+                .into_iter()
+                .map(|line| entry(line, Arc::clone(&default_break))),
+        )
+    }
+
+    fn from_entries(entries: impl IntoIterator<Item = Entry<LineEntry>>) -> Self {
+        let mut work = Work::default();
+        let entries = Sequence::from_entries(entries, &mut work)
+            .expect("validated native line sequence aggregates fit limits");
+        record_sequence(work);
+        Self { entries }
+    }
+
+    pub(super) fn len(&self) -> usize {
+        self.entries.len()
+    }
+
+    pub(super) fn is_empty(&self) -> bool {
+        self.entries.is_empty()
+    }
+
+    pub(super) fn iter(&self) -> Lines<'_> {
+        Lines {
+            entries: self.entries.iter(),
+        }
+    }
+
+    pub(super) fn iter_with_breaks(&self) -> LinesAndBreaks<'_> {
+        LinesAndBreaks { lines: self.iter() }
+    }
+
+    fn get_entry(&self, index: usize) -> Option<&LineEntry> {
+        record(LinePlanWork {
+            line_lookups: 1,
+            ..LinePlanWork::default()
+        });
+        self.entries.get(index).map(|entry| entry.value().as_ref())
+    }
+
+    pub(super) fn get(&self, index: usize) -> Option<&Line> {
+        self.get_entry(index).map(|entry| &entry.line)
+    }
+
+    pub(super) fn first(&self) -> Option<&Line> {
+        self.get(0)
+    }
+
+    pub(super) fn max_width(&self) -> i64 {
+        self.entries.measure().width_max
+    }
+
+    pub(super) fn height(&self) -> i64 {
+        i64::try_from(self.entries.measure().lines).expect("validated native line height fits i64")
+    }
+
+    pub(super) fn break_after(&self, index: usize) -> Option<&AtomProperties> {
+        (index < self.len().saturating_sub(1)).then(|| {
+            self.get_entry(index)
+                .expect("checked line index")
+                .break_after
+                .as_ref()
+        })
+    }
+
+    /// Character offset of a line start, or the complete output length at len().
+    pub(super) fn prefix_chars(&self, end: usize) -> u64 {
+        let chars = self.range_measure(0..end).chars;
+        chars
+            .checked_add(u64::from(end > 0 && end < self.len()))
+            .expect("validated native line prefix fits u64")
+    }
+
+    /// Exact standalone range extent: height, widths, and chars excluding its final newline.
+    pub(super) fn range_measure(&self, range: Range<usize>) -> Measure {
+        record(LinePlanWork {
+            range_queries: 1,
+            ..LinePlanWork::default()
+        });
+        let mut measure = self
+            .entries
+            .range_measure(range)
+            .expect("validated native line range");
+        measure.chars = measure.chars.saturating_sub(u64::from(measure.lines > 0));
+        measure
+    }
+
+    pub(super) fn slice(&self, range: Range<usize>) -> Self {
+        let mut work = Work::default();
+        let entries = self
+            .entries
+            .slice(range, &mut work)
+            .expect("validated native line slice");
+        record_sequence(work);
+        Self { entries }
+    }
+
+    pub(super) fn replace_line(&self, index: usize, line: Line) -> Self {
+        let old = self
+            .get_entry(index)
+            .expect("validated native line replacement");
+        self.replace_entry(index, entry(line, Arc::clone(&old.break_after)))
+    }
+
+    pub(super) fn update_line(&self, index: usize, update: impl FnOnce(Line) -> Line) -> Self {
+        let line = clone_line(self.get(index).expect("validated native line update"));
+        record(LinePlanWork {
+            lines_mapped: 1,
+            ..LinePlanWork::default()
+        });
+        self.replace_line(index, update(line))
+    }
+
+    pub(super) fn with_break(&self, index: usize, properties: AtomProperties) -> Self {
+        let old = self
+            .break_after(index)
+            .expect("validated native break index");
+        if *old == properties {
+            return self.clone();
+        }
+        self.set_stored_break(index, Arc::new(properties))
+    }
+
+    fn set_stored_break(&self, index: usize, properties: Arc<AtomProperties>) -> Self {
+        let old = self
+            .get_entry(index)
+            .expect("validated native break replacement");
+        self.replace_entry(index, entry(clone_line(&old.line), properties))
+    }
+
+    fn replace_entry(&self, index: usize, entry: Entry<LineEntry>) -> Self {
+        let mut work = Work::default();
+        let entries = self
+            .entries
+            .replace(index, entry, &mut work)
+            .expect("validated native line replacement aggregates fit limits");
+        record_sequence(work);
+        Self { entries }
+    }
+
+    pub(super) fn concat(&self, other: &Self) -> Self {
+        self.concat_with_break(other, AtomProperties::default())
+    }
+
+    pub(super) fn concat_with_break(&self, other: &Self, properties: AtomProperties) -> Self {
+        if self.is_empty() {
+            return other.clone();
+        }
+        if other.is_empty() {
+            return self.clone();
+        }
+        let mut work = Work::default();
+        let entries = self
+            .entries
+            .concat(&other.entries, &mut work)
+            .expect("validated native line concatenation aggregates fit limits");
+        record_sequence(work);
+        Self { entries }.with_break(self.len() - 1, properties)
+    }
+
+    /// Full O(N) line transform. Atom payloads remain shared until changed by the callback.
+    pub(super) fn map_lines(&self, mut map: impl FnMut(usize, Line) -> Line) -> Self {
+        let mut iter = self.entries.iter();
+        let result = Self::from_entries(iter.by_ref().enumerate().map(|(index, old)| {
+            record(LinePlanWork {
+                lines_visited: 1,
+                lines_mapped: 1,
+                ..LinePlanWork::default()
+            });
+            entry(
+                map(index, clone_line(&old.value().line)),
+                Arc::clone(&old.value().break_after),
+            )
+        }));
+        record(LinePlanWork {
+            iterator_nodes_visited: iter.nodes_visited(),
+            ..LinePlanWork::default()
+        });
+        result
+    }
+
+    /// Full O(N) break transform. Its traversal and rewritten sequence entries are counted.
+    pub(super) fn map_breaks(&self, mut map: impl FnMut(&mut AtomProperties)) -> Self {
+        let mut iter = self.entries.iter();
+        let last = self.len().saturating_sub(1);
+        let result = Self::from_entries(iter.by_ref().enumerate().map(|(index, old)| {
+            record(LinePlanWork {
+                lines_visited: 1,
+                ..LinePlanWork::default()
+            });
+            if index == last {
+                return old.clone();
+            }
+            let mut properties = old.value().break_after.as_ref().clone();
+            map(&mut properties);
+            record(LinePlanWork {
+                breaks_mapped: 1,
+                ..LinePlanWork::default()
+            });
+            entry(clone_line(&old.value().line), Arc::new(properties))
+        }));
+        record(LinePlanWork {
+            iterator_nodes_visited: iter.nodes_visited(),
+            ..LinePlanWork::default()
+        });
+        result
+    }
+
+    pub(super) fn clear_breaks(&self) -> Self {
+        self.map_breaks(|properties| *properties = AtomProperties::default())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::{MeasuredCluster, Rendered};
+    use super::*;
+
+    fn text(value: &str, width: i64) -> Line {
+        Line::from_clusters(&[MeasuredCluster {
+            text: value.to_owned(),
+            width,
+            cjk: false,
+            space: false,
+            pixel_space: false,
+            source_template_id: None,
+        }])
+    }
+
+    #[test]
+    fn prefix_and_range_summaries_exclude_exactly_one_terminal_newline() {
+        let empty = LinePlan::default();
+        assert_eq!(empty.prefix_chars(0), 0);
+        assert_eq!(empty.range_measure(0..0), Measure::default());
+        let lines = LinePlan::from_lines([text("界a", 3), Line::default(), text("z", 0)]);
+        assert_eq!((lines.height(), lines.max_width()), (3, 3));
+        assert_eq!(
+            (0..=3)
+                .map(|end| lines.prefix_chars(end))
+                .collect::<Vec<_>>(),
+            [0, 3, 4, 5]
+        );
+        let middle = lines.range_measure(1..3);
+        assert_eq!((middle.chars, middle.lines, middle.width_max), (2, 2, 0));
+        let single_empty = lines.slice(1..2);
+        assert_eq!(single_empty.prefix_chars(1), 0);
+        assert_eq!(single_empty.height(), 1);
+        let two_empty = LinePlan::from_lines([Line::default(), Line::default()]);
+        assert_eq!(two_empty.prefix_chars(1), 1);
+        assert_eq!(two_empty.prefix_chars(2), 1);
+        assert_eq!(two_empty.range_measure(0..2).chars, 1);
+    }
+
+    #[test]
+    fn sliced_terminal_break_never_reappears_as_a_default_join() {
+        let original = LinePlan::from_lines([text("a", 1), text("b", 1), text("c", 1)])
+            .with_break(
+                0,
+                AtomProperties {
+                    owner: Some(11),
+                    ..AtomProperties::default()
+                },
+            )
+            .with_break(
+                1,
+                AtomProperties {
+                    scroll_window: Some(12),
+                    ..AtomProperties::default()
+                },
+            );
+        let prefix = original.slice(0..1);
+        assert!(prefix.break_after(0).is_none());
+        let joined = prefix.concat(&original.slice(1..3));
+        assert_eq!(joined.break_after(0), Some(&AtomProperties::default()));
+        assert_eq!(joined.break_after(1).unwrap().scroll_window, Some(12));
+        assert_eq!(original.break_after(0).unwrap().owner, Some(11));
+        let explicit = prefix.concat_with_break(
+            &original.slice(1..3),
+            AtomProperties {
+                owner: Some(13),
+                ..AtomProperties::default()
+            },
+        );
+        assert_eq!(explicit.break_after(0).unwrap().owner, Some(13));
+        assert_eq!(prefix.concat(&LinePlan::default()).prefix_chars(1), 1);
+    }
+
+    #[test]
+    fn replacement_slice_and_concat_share_paths_across_sizes() {
+        for size in [32_usize, 128, 512, 8192] {
+            let original = LinePlan::from_lines((0..size).map(|_| text("a", 1)));
+            let suffix_entry = Arc::clone(original.entries.get(size - 1).unwrap().value());
+            let before = Rendered::from_line_plan(original.clone()).into_tape(0);
+            reset_work();
+            let fork = original.clone();
+            assert_eq!(work(), LinePlanWork::default());
+            let updated = original.replace_line(1, text("hello", 5));
+            let replaced = work();
+            assert!(replaced.sequence_nodes_created <= u64::from(size.ilog2()) + 1);
+            assert!(replaced.sequence_entries_written <= 16);
+            assert_eq!((replaced.lines_mapped, replaced.materialized_lines), (0, 0));
+            assert!(Arc::ptr_eq(
+                updated.entries.get(size - 1).unwrap().value(),
+                &suffix_entry
+            ));
+            assert_eq!(
+                updated.prefix_chars(size - 1),
+                original.prefix_chars(size - 1) + 4
+            );
+            assert_eq!((updated.height(), updated.max_width()), (size as i64, 5));
+            reset_work();
+            let selected = updated.slice(size / 4..size * 3 / 4);
+            let rejoined = updated
+                .slice(0..size / 4)
+                .concat(&selected)
+                .concat(&updated.slice(size * 3 / 4..size));
+            let maintained = work();
+            assert!(maintained.sequence_nodes_created <= 12 * u64::from(size.ilog2()) + 16);
+            assert!(maintained.sequence_entries_written <= 128);
+            assert_eq!(
+                (
+                    maintained.lines_visited,
+                    maintained.lines_mapped,
+                    maintained.materialized_lines
+                ),
+                (0, 0, 0)
+            );
+            assert!(Arc::ptr_eq(
+                rejoined.entries.get(size - 1).unwrap().value(),
+                &suffix_entry
+            ));
+            assert_eq!(Rendered::from_line_plan(fork).into_tape(0), before);
+            let mut oracle = before;
+            oracle.lines[1] = Rendered::from_lines(vec![text("hello", 5)])
+                .into_tape(0)
+                .lines
+                .remove(0);
+            oracle.lines[1].break_after = Some(AtomProperties::default());
+            assert_eq!(Rendered::from_line_plan(rejoined).into_tape(0), oracle);
+        }
+    }
+
+    #[test]
+    fn full_maps_iteration_and_tape_materialization_report_real_work() {
+        let size = 128;
+        let lines = LinePlan::from_lines((0..size).map(|_| text("a", 1)));
+        reset_work();
+        let mapped = lines.map_lines(|_, mut line| {
+            line.push_space(1);
+            line
+        });
+        let line_work = work();
+        assert_eq!(
+            (
+                line_work.lines_visited,
+                line_work.lines_mapped,
+                line_work.line_clones
+            ),
+            (size, size, size)
+        );
+        assert_eq!(line_work.sequence_entries_written, size);
+        assert_eq!(line_work.iterator_nodes_visited, 15);
+        assert_eq!(line_work.materialized_lines, 0);
+        reset_work();
+        let decorated = mapped.map_breaks(|properties| properties.owner = Some(9));
+        let break_work = work();
+        assert_eq!(break_work.lines_visited, size);
+        assert_eq!(break_work.breaks_mapped, size - 1);
+        assert_eq!(break_work.sequence_entries_written, size);
+        assert_eq!(break_work.iterator_nodes_visited, 15);
+        reset_work();
+        let borrowed = decorated.iter().count();
+        assert_eq!(borrowed, size as usize);
+        assert_eq!(work().line_clones, 0);
+        assert_eq!(work().iterator_nodes_visited, 15);
+        reset_work();
+        let tape = Rendered::from_line_plan(decorated).into_tape(0);
+        assert_eq!(work().materialized_lines, size);
+        assert_eq!(work().materialized_breaks, size - 1);
+        assert_eq!(work().materialized_chars, size * 3 - 1);
+        assert_eq!(work().lines_visited, size);
+        assert_eq!(work().iterator_nodes_visited, 15);
+        assert_eq!(work().line_clones, 0);
+        assert!(tape.lines.last().unwrap().break_after.is_none());
+    }
+
+    #[test]
+    fn bulk_vertical_stack_shares_children_with_linear_sequence_work() {
+        for size in [32_usize, 128, 512, 8192] {
+            let children = (0..size)
+                .map(|_| LinePlan::from_lines([text("a", 1)]))
+                .collect::<Vec<_>>();
+            let parts = children
+                .iter()
+                .cloned()
+                .map(Rendered::from_line_plan)
+                .collect();
+            reset_work();
+            let stacked = super::super::stack_vertical(parts);
+            let joined = work();
+            assert_eq!(joined.sequence_nodes_created, size as u64 - 1);
+            assert_eq!(joined.sequence_entries_written, size as u64 * 4);
+            assert_eq!(
+                (
+                    joined.lines_visited,
+                    joined.line_clones,
+                    joined.materialized_lines
+                ),
+                (0, 0, 0)
+            );
+            for (index, child) in children.iter().enumerate() {
+                assert!(Arc::ptr_eq(
+                    stacked.lines.entries.get(index).unwrap().value(),
+                    child.entries.get(0).unwrap().value()
+                ));
+            }
+            assert_eq!(stacked.lines.prefix_chars(size), size as u64 * 2 - 1);
+        }
+    }
+}

native/src/sequence.rs

@@ -1,3 +1,4 @@
+use std::ops::Range;
 use std::sync::Arc;
 
 const LEAF_CAPACITY: usize = 16;
@@ -162,6 +163,51 @@ pub struct Sequence<T> {
     root: Option<Arc<Node<T>>>,
 }
 
+/// Borrowed in-order traversal with an O(log N) stack and no entry clones.
+pub struct Iter<'a, T> {
+    pending: Vec<&'a Node<T>>,
+    leaf: std::slice::Iter<'a, Entry<T>>,
+    remaining: usize,
+    nodes_visited: u64,
+}
+
+impl<T> Iter<'_, T> {
+    pub fn nodes_visited(&self) -> u64 {
+        self.nodes_visited
+    }
+}
+
+impl<'a, T> Iterator for Iter<'a, T> {
+    type Item = &'a Entry<T>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            if let Some(entry) = self.leaf.next() {
+                self.remaining -= 1;
+                return Some(entry);
+            }
+            let node = self.pending.pop()?;
+            self.nodes_visited = self
+                .nodes_visited
+                .checked_add(1)
+                .expect("sequence traversal count fits u64");
+            match node {
+                Node::Leaf { entries, .. } => self.leaf = entries.iter(),
+                Node::Branch { left, right, .. } => {
+                    self.pending.push(right);
+                    self.pending.push(left);
+                }
+            }
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        (self.remaining, Some(self.remaining))
+    }
+}
+
+impl<T> ExactSizeIterator for Iter<'_, T> {}
+
 impl<T> Clone for Sequence<T> {
     fn clone(&self) -> Self {
         Self {
@@ -228,6 +274,15 @@ impl<T> Sequence<T> {
         self.root.as_ref()
     }
 
+    pub fn iter(&self) -> Iter<'_, T> {
+        Iter {
+            pending: self.root.as_deref().into_iter().collect(),
+            leaf: [].iter(),
+            remaining: self.len(),
+            nodes_visited: 0,
+        }
+    }
+
     pub fn get(&self, index: usize) -> Option<&Entry<T>> {
         fn get_at<T>(node: &Node<T>, index: usize) -> Option<&Entry<T>> {
             match node {
@@ -270,6 +325,55 @@ impl<T> Sequence<T> {
             .map_or(Ok(Measure::default()), |root| prefix(root, end))
     }
 
+    /// Aggregate a range by visiting its boundary paths and cached interior nodes.
+    pub fn range_measure(&self, range: Range<usize>) -> Result<Measure, Error> {
+        fn measure<T>(node: &Node<T>, start: usize, end: usize) -> Result<Measure, Error> {
+            if start == end {
+                return Ok(Measure::default());
+            }
+            if start == 0 && end == node.len() {
+                return Ok(node.measure());
+            }
+            match node {
+                Node::Leaf { entries, .. } => entries[start..end]
+                    .iter()
+                    .try_fold(Measure::default(), |sum, entry| {
+                        sum.checked_combine(entry.measure)
+                    }),
+                Node::Branch { left, right, .. } => {
+                    let middle = left.len();
+                    if end <= middle {
+                        measure(left, start, end)
+                    } else if start >= middle {
+                        measure(right, start - middle, end - middle)
+                    } else {
+                        measure(left, start, middle)?.checked_combine(measure(
+                            right,
+                            0,
+                            end - middle,
+                        )?)
+                    }
+                }
+            }
+        }
+        if range.start > range.end || range.end > self.len() {
+            return Err(Error::IndexOutOfBounds);
+        }
+        self.root.as_deref().map_or(Ok(Measure::default()), |root| {
+            measure(root, range.start, range.end)
+        })
+    }
+
+    /// Share the interior of a range, copying only its two boundary paths.
+    pub fn slice(&self, range: Range<usize>, work: &mut Work) -> Result<Self, Error> {
+        if range.start > range.end || range.end > self.len() {
+            return Err(Error::IndexOutOfBounds);
+        }
+        let (before, _) = self.split_at(range.end, work)?;
+        let (_, selected) = before.split_at(range.start, work)?;
+        Ok(selected)
+    }
+
     pub fn split_at(&self, index: usize, work: &mut Work) -> Result<(Self, Self), Error> {
         if index > self.len() {
             return Err(Error::IndexOutOfBounds);
@@ -642,6 +746,47 @@ mod tests {
         }
     }
 
+    #[test]
+    fn borrowed_iteration_and_range_operations_match_exact_measures() {
+        let mut setup = Work::default();
+        let original = Sequence::from_entries((0..97).map(entry), &mut setup).unwrap();
+        let mut iter = original.iter();
+        for index in 0..original.len() {
+            assert_eq!(iter.len(), original.len() - index);
+            assert!(std::ptr::eq(
+                iter.next().unwrap(),
+                original.get(index).unwrap()
+            ));
+        }
+        assert!(iter.next().is_none());
+        assert_eq!(iter.nodes_visited(), setup.nodes_created);
+        for start in 0..=97 {
+            for end in start..=97 {
+                let expected = original
+                    .iter()
+                    .skip(start)
+                    .take(end - start)
+                    .try_fold(Measure::default(), |sum, entry| {
+                        sum.checked_combine(entry.measure())
+                    })
+                    .unwrap();
+                assert_eq!(original.range_measure(start..end).unwrap(), expected);
+                let mut work = Work::default();
+                let selected = original.slice(start..end, &mut work).unwrap();
+                assert_eq!(selected.measure(), expected);
+                assert!(selected
+                    .iter()
+                    .zip(original.iter().skip(start))
+                    .all(|(a, b)| Arc::ptr_eq(a.value(), b.value())));
+            }
+        }
+        assert_eq!(
+            original.slice(98..98, &mut Work::default()).unwrap_err(),
+            Error::IndexOutOfBounds
+        );
+        assert_eq!(original.range_measure(98..98), Err(Error::IndexOutOfBounds));
+    }
+
     #[test]
     fn repeated_left_and_right_concat_remain_balanced() {
         let mut left = Sequence::empty();