From 97b7be20ed0f39f15abc737cc1cf0d719e8e33f7 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 14:09:28 -0400
Subject: [PATCH 1/7] Checkpoint native participant capture; real-history frame
 pacing remains unqualified

Capture borrows exact original native paint participants and resource identities instead of repeated whole-DOM scans. Current retirement waits through the native publication owner; preceding-source publication still captures synchronously. Preserves exposure, style, geometry and source custody. Related capture consumers migrated, with parked positive retirement fixture moved genuinely offscreen.

Exposure check passed and all three body families reached original retained rows, reentry, style and pending-writer checks. Full fixture still refused the final immediate post-stream retirement; source owner currentness must settle. Real 320-packet saved-history run at original147260275 frontier regressed to866ms maximum versus393ms baseline, App errorNone; this checkpoint is not delivered or a performance pass. Existing negative outputs retained under dm-native-caption-20261008.
---
 src/toad/widgets/viewport_body.py            | 189 ++++++++++++-------
 tests/retained_fragment_geometry_pilot.py    |  31 ++-
 tests/retained_tool_text_pilot.py            |   3 +-
 tests/test_worker_preparation_acquisition.py |   6 +-
 tests/viewport_body_lifetime_pilot.py        |   9 +-
 5 files changed, 155 insertions(+), 83 deletions(-)

diff --git a/src/toad/widgets/viewport_body.py b/src/toad/widgets/viewport_body.py
index 22609c5ac..4dfb29430 100644
--- a/src/toad/widgets/viewport_body.py
+++ b/src/toad/widgets/viewport_body.py
@@ -222,9 +222,12 @@ def requires_geometry(self, body) -> bool:
     def geometry_targets(self, body):
         return (body,) if self.requires_geometry(body) else ()
 
-    def capture_requested(self):
+    def capture_requested(self, targets=None):
         return self
 
+    def capture_preparation(self, body, targets):
+        """Source publication does not acquire a retirement geometry demand."""
+
     def required(self):
         return self
 
@@ -294,13 +297,16 @@ class LiveBody(MeasuredBody):
     def released(self):
         return ReleasedBody(self.width, self.rows, self.widgets)
 
-    def capture_requested(self):
-        return CapturingBody(self.width, self.rows, self.widgets)
+    def capture_requested(self, targets=None):
+        return CapturingBody(self.width, self.rows, self.widgets,
+                             targets=() if targets is None else targets)
 
     def prepare_publication(self, body):
         # Borrow the original published scene synchronously, before a writer
         # or its byte measurement can change the native tree.
-        return body.capture_native_paint(self)
+        capture = body.capture_native_paint(self)
+        return (capture.prepare(body, self) if capture is not None
+                else MeasuredBody(self.width, self.rows, self.widgets).prepare_publication(body))
 
     def invalidated(self):
         # The original native tree still carries its pixels. Replace the
@@ -335,6 +341,16 @@ async def materialize(self, body):
     def retire(self, body):
         return body.retire_native_body(self)
 
+    def capture_retirement(self, body, children):
+        # Retirement acquires the current scene after the native frame owner
+        # publishes its geometry. Preceding-source capture above must instead
+        # borrow synchronously before that source's writer changes the tree.
+        capture = body.capture_native_paint(self)
+        if capture is not None:
+            return capture.retire(body, self, children)
+        body._update_body_measurement(self.capture_requested())
+        return BodyMeasurement.retire(self, body)
+
 
 @dataclass(frozen=True)
 class CapturingBody(LiveBody):
@@ -347,12 +363,22 @@ class CapturingBody(LiveBody):
     """
 
     capture_pending = True
+    targets: tuple[Widget, ...] = ()
 
-    def capture_requested(self):
-        return self
+    def capture_requested(self, targets=None):
+        return self if targets is None or targets == self.targets else replace(self, targets=targets)
+
+    def capture_preparation(self, body, targets):
+        if body._body_measurement is self:
+            body._update_body_measurement(self.capture_requested(targets))
 
     def geometry_targets(self, body):
-        return (body, *walk_depth_first(body, PreparedPaintSource, with_root=False))
+        return (body, *self.targets)
+
+    def capture_retirement(self, body, children):
+        capture = body.capture_native_paint(self)
+        return (capture.retire(body, self, children) if capture is not None
+                else BodyMeasurement.retire(self, body))
 
     def required(self):
         return LiveBody(self.width, self.rows, self.widgets)
@@ -589,6 +615,51 @@ async def restore(self, body):
             await body.materialize_body()
 
 
+@dataclass(frozen=True)
+class BodyPaintCapture:
+    """Borrowed native paint and its participants until preparation/pruning.
+
+    Compositor supplies this exact cohort once. The temporary acquisition owns
+    its original source witnesses; the retained body owns only immutable rows,
+    so pruning does not keep the retired child widgets and resources alive.
+    """
+
+    content: PreparedRichContent
+    sources: tuple[tuple[PreparedPaintSource, PreparedRichContent], ...]
+    bodies: tuple[ViewportBody, ...]
+    style_revision: int
+    geometry_revision: int
+    paint_state: PaintState
+    widgets: int
+
+    def current(self, body):
+        return (self.style_revision == body._subtree_style_revision
+                and self.geometry_revision == body._geometry_revision
+                and all(owner.is_attached and owner.prepared_content is resource
+                        for owner, resource in self.sources)
+                and all(owner.is_attached and owner.body_ready for owner in self.bodies))
+
+    async def prepare(self, body, current):
+        size_bytes = await body.app.preparation.run_thread(retained_bytes, self.content)
+        return RenderedBody(
+            current.width, current.rows, self.widgets, content=self.content,
+            resource_bytes=size_bytes, style_revision=self.style_revision, paint_state=self.paint_state,
+        )
+
+    async def retire(self, body, current, children):
+        rendered = await self.prepare(body, current)
+        if not body.retirement_current(current) or not self.current(body) or not rendered.ready(body):
+            return False
+        async with body.retirement_custody() as can_commit:
+            if (not can_commit or not body.retirement_current(current)
+                    or not self.current(body) or not rendered.ready(body)):
+                return False
+            body._update_body_measurement(rendered)
+            body.retire_body_resources()
+            await body.remove_children(children)
+        return True
+
+
 class MeasuredViewportBody(ViewportBody):
     """The body owns Live, Rendered and Measured behavior, not copied flags."""
 
@@ -680,6 +751,9 @@ def _update_body_measurement(self, measurement):
         # NodeList removal happens later and cannot invalidate it for us.
         self.refresh(layout=True)
         if self._body_viewport is not None:
+            if measurement.capture_pending:
+                self.screen.frame_presentation.defer(
+                    self._body_viewport.window, self._body_viewport.request)
             self._body_viewport.request()
 
     def notify_style_update(self):
@@ -763,65 +837,52 @@ def retire_native_body(self, current):
         children = self.reconstructible_children()
         if not children:
             return BodyMeasurement.retire(current, self)
-        sources = self.prepared_paint_sources()
-        if not self.prepared_paint_is_current(sources):
+        screen = self.screen
+        if self._after_refresh_pending(screen, screen._layout_mutation_roots()):
             self._update_body_measurement(current.capture_requested())
+            screen.frame_presentation.defer(self._body_viewport.window, self._body_viewport.request)
             return BodyMeasurement.retire(current, self)
-        paint = self.capture_native_paint(current, sources=sources)
-        return self.finish_native_retirement(current, children, paint, sources)
-
-    def prepared_paint_sources(self):
-        """Borrow original committed resources, including nested body contents."""
-        return tuple((child, child.prepared_content)
-                     for child in walk_depth_first(self, with_root=True)
-                     if isinstance(child, PreparedPaintSource))
-
-    def prepared_paint_is_current(self, sources):
-        current = self.prepared_paint_sources()
-        return (len(current) == len(sources)
-                and all(owner is original and resource is not None and resource is prepared
-                        for (owner, resource), (original, prepared) in zip(current, sources))
-                and all(child.body_ready for child in walk_depth_first(self, with_root=False)
-                        if isinstance(child, ViewportBody)))
-
-    def capture_native_paint(self, current, *, sources=None):
+        return current.capture_retirement(self, children)
+
+    def capture_native_paint(self, current):
         """Capture once for retirement and preceding-source publication."""
-        if sources is None:
-            sources = self.prepared_paint_sources()
-        unprepared = MeasuredBody(current.width, current.rows, current.widgets)
         if (not current.ready(self) or not self.is_attached or self.lock.is_locked
                 or any(self in endpoint.walk_ancestors(with_self=True)
-                       for endpoint in self.screen._interaction_widgets())
-                or not self.prepared_paint_is_current(sources)):
-            return unprepared.prepare_publication(self)
+                       for endpoint in self.screen._interaction_widgets())):
+            return None
         compositor = self.screen._compositor
-        style_revision = self._subtree_style_revision
-        paint_state = self._resolved_paint_state()
         captured = tuple(compositor.published_geometry((self,)))
         if not captured:
-            return unprepared.prepare_publication(self)
+            return None
         _body, placement = captured[0]
-        size, rows = compositor.render_subtree_strips(self, placement)
-        # Arrangement may acquire a different width and start preparation.
-        # Its provisional rows cannot become the source's retained resource.
-        if not self.prepared_paint_is_current(sources):
-            return unprepared.prepare_publication(self)
+        sources = ()
+        bodies = ()
+
+        def admit(participants):
+            nonlocal sources, bodies
+            sources = tuple((owner, owner.prepared_content) for owner in participants
+                            if isinstance(owner, PreparedPaintSource))
+            bodies = tuple(owner for owner in participants
+                           if owner is not self and isinstance(owner, ViewportBody))
+            if (any(resource is None for _, resource in sources)
+                    or any(not owner.body_ready for owner in bodies)):
+                current.capture_preparation(self, tuple(dict.fromkeys(
+                    (*bodies, *(owner for owner, _ in sources)))))
+                return False
+            return True
+
+        native = compositor.render_subtree_strips(self, placement, admit=admit)
+        if native is None:
+            return None
+        size, rows = native
         # Native capture already owns final styled terminal rows. Retain those
         # rows directly; rendering them again in Rich workers duplicates work
         # and serializes a resource that never leaves this process.
         content = PreparedRichContent(size.width, tuple(rows))
-        widgets = self.materialized_widget_count
-
-        async def measured():
-            size_bytes = await self.app.preparation.run_thread(retained_bytes, content)
-            if not self.prepared_paint_is_current(sources):
-                return unprepared
-            return RenderedBody(
-                current.width, current.rows, widgets, content=content,
-                resource_bytes=size_bytes, style_revision=style_revision, paint_state=paint_state,
-            )
-
-        return measured()
+        return BodyPaintCapture(
+            content, sources, bodies, self._subtree_style_revision,
+            self._geometry_revision, self._resolved_paint_state(), self.materialized_widget_count,
+        )
 
     def retirement_current(self, current):
         """Borrow current viewport demand again after asynchronous capture."""
@@ -843,23 +904,6 @@ async def release_native_body(self, current):
             await self.remove_children(children)
         return True
 
-    async def finish_native_retirement(self, current, children, paint, sources):
-        rendered = await paint
-        # The original state is the captured source/width/style lifetime. An
-        # asynchronous change invalidates it rather than copying a revision.
-        if not self.retirement_current(current):
-            return False
-        if not rendered.ready(self):
-            return False
-        async with self.retirement_custody() as can_commit:
-            if (not can_commit or not self.retirement_current(current) or not rendered.ready(self)
-                    or not self.prepared_paint_is_current(sources)):
-                return False
-            self._update_body_measurement(rendered)
-            self.retire_body_resources()
-            await self.remove_children(children)
-        return True
-
     async def restore_body(self):
         previous = self._body_measurement
         await previous.restore(self)
@@ -1403,6 +1447,11 @@ async def _reconcile(self) -> None:
                         return
                 if self._pending:
                     continue
+                if screen.frame_presentation.awaits_publication(self.window, self.request):
+                    # Foreground restoration remains free to prepare all
+                    # exposed siblings. Retirement resumes from the original
+                    # native callback after their geometry and paint publish.
+                    return
                 retiring = tuple(owner for owner in owners
                                  if owner.is_attached and not owner._closing
                                  and owner not in retained and not owner.body_dormant)
diff --git a/tests/retained_fragment_geometry_pilot.py b/tests/retained_fragment_geometry_pilot.py
index 44ec6dacd..5f4f11a90 100644
--- a/tests/retained_fragment_geometry_pilot.py
+++ b/tests/retained_fragment_geometry_pilot.py
@@ -16,6 +16,7 @@
 from agent_comms.comms import Comms
 from textual._compositor import Compositor
 from toad.app import ToadApp
+from toad.rich_preparation import PreparedPaintSource
 from toad.widgets.transcript_history import TranscriptHistory
 
 
@@ -63,11 +64,16 @@ async def settle():
             async def acquire_capture(member):
                 """Exercise the body's original complete-row acquisition."""
                 async with asyncio.timeout(10):
-                    while not member.prepared_paint_is_current(member.prepared_paint_sources()):
+                    while True:
+                        await pilot.pause(.02)
+                        if not member._after_refresh_pending(app.screen, app.screen._layout_mutation_roots()):
+                            capture = member.capture_native_paint(member._body_measurement)
+                            if (capture is not None and capture.current(member)
+                                    and not member._after_refresh_pending(app.screen, app.screen._layout_mutation_roots())):
+                                return
                         if member.body_ready and not member.body_capture_pending:
                             assert not await member.retire_body()
                             assert member.body_capture_pending
-                        await pilot.pause(.02)
 
             await settle()
             live_targets = tuple(owner for owner in viewport.body_roots()
@@ -154,7 +160,17 @@ def trace(frame, event, arg):
             await pilot.pause()
             compare()
             receipt['resize_invalidated'] = True
+            # Native exposure owns source custody even in this parked fixture.
+            # The styled fragment is visible here and may not be pruned.
+            assert not await body.retire_body()
+            window.styles.height = 4
+            await pilot.pause()
+            window.release_anchor()
+            window.scroll_to(y=window.max_scroll_y, animate=False, immediate=True)
+            await pilot.pause()
+            assert not viewport.requires_body(body)
             old_children = set(body.walk_children())
+            await acquire_capture(body)
             assert await body.retire_body()
             await pilot.pause()
             retired_scene, _ = compare()
@@ -167,7 +183,10 @@ def trace(frame, event, arg):
             # A restoration may not parse/remount at the same width.
             family = [body, PreparedConversationMarkdown("# Native prepared body\n\nRetained full lines."),
                       StreamingMarkdown("# Native streaming body\n\nRetained full lines.", paginate=False)]
-            await window.mount(*family[1:])
+            from textual.widget import Widget
+            trailing_space = Widget()
+            trailing_space.styles.height = 40
+            await window.mount(*family[1:], trailing_space)
             await pilot.pause()
             window.release_anchor()
             window.scroll_to(y=window.max_scroll_y, animate=False, immediate=True)
@@ -225,7 +244,7 @@ def observe_capture(frame, event, arg):
                 published_scene = scene._full_map, scene._visible_map
                 captured_member, placement = next(scene.published_geometry((member,)))
                 assert captured_member is member
-                scene.render_subtree_strips(member, placement)
+                scene.render_subtree_strips(member, placement, admit=lambda participants: True)
                 assert scene._full_map is published_scene[0]
                 assert scene._visible_map is published_scene[1]
                 if not member.body_dormant:
@@ -402,7 +421,8 @@ async def second_source():
                     member.lock.is_locked, member in viewport.protected(),
                     [(type(source).__name__, source.presentation_ready,
                       source.prepared_content is not None, tuple(source.size))
-                     for source, _ in member.prepared_paint_sources()])
+                     for source in member.body_geometry_targets()
+                     if isinstance(source, PreparedPaintSource)])
                 await pilot.pause()
                 family_receipts[-1]['settled_preceding_writer_released_without_replacing_current_paint'] = True
             receipt['rendered_family_reentry'] = family_receipts
@@ -420,6 +440,7 @@ async def second_source():
             assert not await operation
             assert not streaming.body_dormant
             await streaming.finish_stream()
+            await acquire_capture(streaming)
             assert await streaming.retire_body()
             await pilot.pause()
             receipt['inflight_stream_blocks_captured_retirement_commit'] = True
diff --git a/tests/retained_tool_text_pilot.py b/tests/retained_tool_text_pilot.py
index 350071227..f390099a3 100644
--- a/tests/retained_tool_text_pilot.py
+++ b/tests/retained_tool_text_pilot.py
@@ -97,8 +97,7 @@ async def publication_lifetime(app, pilot, tool):
     assert text in set(app.screen._interaction_widgets())
     assert body in window.document_viewport.protected()
     assert tool in protected_blocks(conversation, (tool,))
-    measured = await body.capture_native_paint(body._body_measurement)
-    assert not isinstance(measured, RenderedBody)
+    assert body.capture_native_paint(body._body_measurement) is None
     await tool.update_tool_call(payload("gesture replacement"))
     assert body.query_one(TextContent) is not text
     assert app.screen._select_state is None and text not in app.screen.selections
diff --git a/tests/test_worker_preparation_acquisition.py b/tests/test_worker_preparation_acquisition.py
index d485e1cad..3c43791fc 100644
--- a/tests/test_worker_preparation_acquisition.py
+++ b/tests/test_worker_preparation_acquisition.py
@@ -662,10 +662,10 @@ async def mounted():
                         "content_lock": document._content_lock.locked(),
                         "stream": str(document._stream),
                         "children": len(document.reconstructible_children()),
-                        "prepared_current": document.prepared_paint_is_current(document.prepared_paint_sources()),
+                        "capture_admitted": document.capture_native_paint(document._body_measurement) is not None,
                         "published_placement": str(tuple(app.screen._compositor.published_geometry((document,)))),
-                        "missing_paint": [str(owner) for owner, content in document.prepared_paint_sources()
-                                          if content is None],
+                        "missing_paint": [str(owner) for owner in document.body_geometry_targets()
+                                          if isinstance(owner, PreparedPaintSource) and owner.prepared_content is None],
                         "nested_not_ready": [str(owner) for owner in walk_depth_first(document, MeasuredViewportBody)
                                              if not owner.body_ready],
                     },
diff --git a/tests/viewport_body_lifetime_pilot.py b/tests/viewport_body_lifetime_pilot.py
index fa23dfff2..8c7effe08 100644
--- a/tests/viewport_body_lifetime_pilot.py
+++ b/tests/viewport_body_lifetime_pilot.py
@@ -16,6 +16,7 @@
 
 from runtime_fixture import ToadApp
 from toad.widgets.agent_response import AgentResponse
+from toad.rich_preparation import PreparedPaintSource
 
 
 async def settled(view, pilot):
@@ -35,7 +36,9 @@ async def settled(view, pilot):
         if output:
             roots = []
             for owner in manager.body_roots():
-                sources = owner.prepared_paint_sources()
+                sources = tuple((child, child.prepared_content)
+                                for child in owner.body_geometry_targets()
+                                if isinstance(child, PreparedPaintSource))
                 roots.append(dict(
                     measurement=type(owner._body_measurement).__name__,
                     ready=owner.body_ready, required=manager.requires_body(owner),
@@ -164,7 +167,7 @@ async def retirement_visibility(output: Path):
             first = AgentResponse("Original body\n\nActual retained paragraph.")
             await view.contents.mount(first)
             async with asyncio.timeout(15):
-                while not first.body_ready or not first.prepared_paint_is_current(first.prepared_paint_sources()):
+                while not first.body_ready:
                     await pilot.pause(.02)
             # Suspend original housekeeping so this check owns one retirement.
             await manager.suspend_source()
@@ -174,7 +177,7 @@ async def retirement_visibility(output: Path):
             await pilot.pause()
             assert not manager.requires_body(first)
             children = first.reconstructible_children()
-            assert children and first.prepared_paint_is_current(first.prepared_paint_sources())
+            assert children and first.capture_native_paint(first._body_measurement) is not None
             operation = first.retire_body()
             # Actual scrolling exposes the body after synchronous acquisition,
             # before the returned operation measures/commits its captured rows.

From bc67d486b64d4a529ce7f72261af094792ae1906 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 16:00:48 -0400
Subject: [PATCH 2/7] Derive body layout demand from extent and native
 participation

---
 src/toad/widgets/viewport_body.py | 14 ++++++++++++--
 1 file changed, 12 insertions(+), 2 deletions(-)

diff --git a/src/toad/widgets/viewport_body.py b/src/toad/widgets/viewport_body.py
index 4dfb29430..168699f66 100644
--- a/src/toad/widgets/viewport_body.py
+++ b/src/toad/widgets/viewport_body.py
@@ -743,13 +743,23 @@ def invalidate_body(self):
         self._update_body_measurement(self._body_measurement.invalidated())
 
     def _update_body_measurement(self, measurement):
-        if measurement is self._body_measurement:
+        previous = self._body_measurement
+        if measurement is previous:
             return
+        # State/resource publication is not itself a new layout. Native extent,
+        # descendant participation and geometry demand own that decision. Byte
+        # accounting and writer identity still repaint/request reconciliation
+        # without arranging every unchanged body in the window again.
+        layout = (
+            (measurement.width, measurement.rows) != (previous.width, previous.rows)
+            or measurement.paint_ready(self) != previous.paint_ready(self)
+            or measurement.geometry_targets(self) != previous.geometry_targets(self)
+        )
         self._body_measurement = measurement
         # The resource owns descendant participation and cover selection as
         # well as extent. Publish that change before any native prune awaits;
         # NodeList removal happens later and cannot invalidate it for us.
-        self.refresh(layout=True)
+        self.refresh(layout=layout)
         if self._body_viewport is not None:
             if measurement.capture_pending:
                 self.screen.frame_presentation.defer(

From 8bce86fca5c3fb2d36d61d9f0e208b5f9a397222 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 16:32:14 -0400
Subject: [PATCH 3/7] Yield retired body painting through original capture
 lifetimes

---
 src/toad/widgets/viewport_body.py         | 66 +++++++++++++++--------
 tests/retained_fragment_geometry_pilot.py | 33 ++++++++++--
 2 files changed, 72 insertions(+), 27 deletions(-)

diff --git a/src/toad/widgets/viewport_body.py b/src/toad/widgets/viewport_body.py
index 168699f66..3877867bf 100644
--- a/src/toad/widgets/viewport_body.py
+++ b/src/toad/widgets/viewport_body.py
@@ -6,7 +6,7 @@
 """
 
 from contextlib import AsyncExitStack, ExitStack, asynccontextmanager
-from collections.abc import Awaitable, Callable, Coroutine
+from collections.abc import Awaitable, Callable, Coroutine, Iterator
 from collections import OrderedDict
 from functools import partial
 from weakref import WeakSet, ref
@@ -624,9 +624,10 @@ class BodyPaintCapture:
     so pruning does not keep the retired child widgets and resources alive.
     """
 
-    content: PreparedRichContent
+    size: Size
+    bands: Iterator[list[Strip]]
     sources: tuple[tuple[PreparedPaintSource, PreparedRichContent], ...]
-    bodies: tuple[ViewportBody, ...]
+    bodies: tuple[tuple[ViewportBody, BodyMeasurement], ...]
     style_revision: int
     geometry_revision: int
     paint_state: PaintState
@@ -637,17 +638,39 @@ def current(self, body):
                 and self.geometry_revision == body._geometry_revision
                 and all(owner.is_attached and owner.prepared_content is resource
                         for owner, resource in self.sources)
-                and all(owner.is_attached and owner.body_ready for owner in self.bodies))
-
-    async def prepare(self, body, current):
-        size_bytes = await body.app.preparation.run_thread(retained_bytes, self.content)
+                and all(owner.is_attached and owner._body_measurement is resource
+                        and resource.ready(owner) for owner, resource in self.bodies))
+
+    def prepare(self, body, current):
+        # A source writer starts as soon as this call returns. Drain its
+        # preceding pixels now, rather than leaving mutable native readers in
+        # the asynchronous byte-preparation task.
+        content = PreparedRichContent(self.size.width, tuple(
+            strip for band in self.bands for strip in band))
+        return self._prepare(body, current, content)
+
+    async def _prepare(self, body, current, content):
+        size_bytes = await body.app.preparation.run_thread(retained_bytes, content)
         return RenderedBody(
-            current.width, current.rows, self.widgets, content=self.content,
+            current.width, current.rows, self.widgets, content=content,
             resource_bytes=size_bytes, style_revision=self.style_revision, paint_state=self.paint_state,
         )
 
     async def retire(self, body, current, children):
-        rendered = await self.prepare(body, current)
+        rows = []
+        while True:
+            # Geometry is lent only inside next(). Input, source publication
+            # and ordinary frames use their own scene between these bands.
+            # Reacquire original lifetimes before every native read, including
+            # after the last yield; stale captures never reach pruning.
+            if not body.retirement_current(current) or not self.current(body):
+                return False
+            try:
+                rows.extend(next(self.bands))
+            except StopIteration:
+                break
+            await asyncio.sleep(0)
+        rendered = await self._prepare(body, current, PreparedRichContent(self.size.width, tuple(rows)))
         if not body.retirement_current(current) or not self.current(body) or not rendered.ready(body):
             return False
         async with body.retirement_custody() as can_commit:
@@ -828,8 +851,8 @@ def retire_body_resources(self):
         pass
 
     def retire_body(self):
-        # Acquire pixels synchronously. The viewport can capture its bounded
-        # cohort from one publication before any preparation or pruning awaits.
+        # Acquire geometry and resource witnesses synchronously. Retirement
+        # paints bands in its original task before byte preparation/pruning.
         return self._body_measurement.retire(self)
 
     @asynccontextmanager
@@ -872,25 +895,24 @@ def admit(participants):
             nonlocal sources, bodies
             sources = tuple((owner, owner.prepared_content) for owner in participants
                             if isinstance(owner, PreparedPaintSource))
-            bodies = tuple(owner for owner in participants
+            bodies = tuple((owner, owner._body_measurement) for owner in participants
                            if owner is not self and isinstance(owner, ViewportBody))
             if (any(resource is None for _, resource in sources)
-                    or any(not owner.body_ready for owner in bodies)):
+                    or any(not resource.ready(owner) for owner, resource in bodies)):
                 current.capture_preparation(self, tuple(dict.fromkeys(
-                    (*bodies, *(owner for owner, _ in sources)))))
+                    (*(owner for owner, _ in bodies), *(owner for owner, _ in sources)))))
                 return False
             return True
 
         native = compositor.render_subtree_strips(self, placement, admit=admit)
         if native is None:
             return None
-        size, rows = native
-        # Native capture already owns final styled terminal rows. Retain those
-        # rows directly; rendering them again in Rich workers duplicates work
-        # and serializes a resource that never leaves this process.
-        content = PreparedRichContent(size.width, tuple(rows))
+        size, bands = native
+        # Witnesses precede the first paint or await. Native bands already
+        # supply styled terminal rows; only detached byte accounting belongs
+        # in preparation workers, not another Rich rendering of those rows.
         return BodyPaintCapture(
-            content, sources, bodies, self._subtree_style_revision,
+            size, bands, sources, bodies, self._subtree_style_revision,
             self._geometry_revision, self._resolved_paint_state(), self.materialized_widget_count,
         )
 
@@ -1474,8 +1496,8 @@ async def _reconcile(self) -> None:
                 retiring = retiring[:self.budget.admission_items]
                 retired_owners = []
                 for first in range(0, len(retiring), self.budget.admission_items):
-                    # Each call captures its original rows now; tasks only
-                    # prepare detached rows and validate/prune afterward.
+                    # Each call acquires the original geometry/resources now;
+                    # tasks paint with lifetime checks between native bands.
                     batch = retiring[first:first + self.budget.admission_items]
                     with ExitStack() as captures:
                         operations = []
diff --git a/tests/retained_fragment_geometry_pilot.py b/tests/retained_fragment_geometry_pilot.py
index 5f4f11a90..5552a058a 100644
--- a/tests/retained_fragment_geometry_pilot.py
+++ b/tests/retained_fragment_geometry_pilot.py
@@ -181,8 +181,9 @@ def trace(frame, event, arg):
             assert body.body_ready and not old_children & set(body.walk_children())
             # The original body family paints full retained rows on reentry.
             # A restoration may not parse/remount at the same width.
-            family = [body, PreparedConversationMarkdown("# Native prepared body\n\nRetained full lines."),
-                      StreamingMarkdown("# Native streaming body\n\nRetained full lines.", paginate=False)]
+            full_body = "\n\n".join(f"Retained native paragraph {index}." for index in range(40))
+            family = [body, PreparedConversationMarkdown("# Native prepared body\n\n" + full_body),
+                      StreamingMarkdown("# Native streaming body\n\n" + full_body, paginate=False)]
             from textual.widget import Widget
             trailing_space = Widget()
             trailing_space.styles.height = 40
@@ -206,6 +207,24 @@ def trace(frame, event, arg):
             for member in family:
                 await acquire_capture(member)
             scene.reflow(app.screen, app.size)
+            # A real declaration write may arrive between native bands. The
+            # original capture must discard its partial rows and leave every
+            # native child in custody, rather than publish mixed-style paint.
+            for member in family:
+                children = tuple(member.children)
+                original_color = member.styles.color
+                retirement = asyncio.create_task(member.retire_body())
+                asyncio.get_running_loop().call_soon(setattr, member.styles, 'color', '#bada55')
+                try:
+                    assert not await retirement, type(member).__name__
+                    assert not member.body_dormant and tuple(member.children) == children
+                    assert all(child.is_attached for child in children)
+                finally:
+                    member.styles.color = original_color
+                await pilot.pause()
+                scene.reflow(app.screen, app.size)
+                await acquire_capture(member)
+            receipt['between_band_style_write_retains_all_native_children'] = len(family)
             native_captures = Counter()
 
             def observe_capture(frame, event, arg):
@@ -227,8 +246,8 @@ def observe_capture(frame, event, arg):
                         operations.append(operation)
                 finally:
                     sys.setprofile(None)
-                # Borrow rows before preparation can yield and first removal
-                # can invalidate sibling placement in the original scene.
+                # Acquire each cohort's original geometry/resources before
+                # band painting or first removal can yield to another task.
                 assert native_captures == Counter(bodies=len(family)), native_captures
                 assert all(not member.body_dormant for member in family)
                 assert all(await asyncio.gather(*operations))
@@ -244,7 +263,11 @@ def observe_capture(frame, event, arg):
                 published_scene = scene._full_map, scene._visible_map
                 captured_member, placement = next(scene.published_geometry((member,)))
                 assert captured_member is member
-                scene.render_subtree_strips(member, placement, admit=lambda participants: True)
+                native = scene.render_subtree_strips(member, placement, admit=lambda participants: True)
+                assert native is not None
+                _, bands = native
+                for _ in bands:
+                    pass
                 assert scene._full_map is published_scene[0]
                 assert scene._visible_map is published_scene[1]
                 if not member.body_dormant:

From de83aa39f87e3ea64262803eae947f5277a67a06 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 17:41:32 -0400
Subject: [PATCH 4/7] Acquire held geometry through history publication and
 retain real layout completion

---
 src/toad/screens/workspace.py                |  3 +-
 src/toad/widgets/history_anchor.py           | 76 ++++++++++++--------
 src/toad/widgets/viewport_body.py            | 10 ++-
 tests/history_reflow_travel_pilot.py         |  2 +-
 tests/retained_fragment_geometry_pilot.py    |  4 +-
 tests/test_worker_preparation_acquisition.py |  2 +
 tests/viewport_body_lifetime_pilot.py        | 11 ++-
 7 files changed, 69 insertions(+), 39 deletions(-)

diff --git a/src/toad/screens/workspace.py b/src/toad/screens/workspace.py
index f371dcdc6..1877fc451 100644
--- a/src/toad/screens/workspace.py
+++ b/src/toad/screens/workspace.py
@@ -20,6 +20,7 @@
 from abc import abstractmethod
 
 if TYPE_CHECKING:
+    from textual._compositor import SubtreeGeometryPlacement
     from toad.app import ToadApp
 
 
@@ -158,7 +159,7 @@ def update_node_styles(self, animate: bool = True) -> None:
             # Record that completed update instead of repeating it next resume.
             self._resume_style = self._style_revision()
 
-    def _layout_mutation_roots(self) -> tuple[Widget, ...]:
+    def _layout_mutation_roots(self) -> dict[Widget, SubtreeGeometryPlacement | None]:
         return self.viewport_presentation.mutation_roots()
 
     def _prepare_compositor_refresh(self) -> tuple[Widget, ...]:
diff --git a/src/toad/widgets/history_anchor.py b/src/toad/widgets/history_anchor.py
index cd79de943..b0e4cff22 100644
--- a/src/toad/widgets/history_anchor.py
+++ b/src/toad/widgets/history_anchor.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
+from dataclasses import dataclass, field, replace
 from functools import cached_property
 from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
 import asyncio
@@ -18,6 +18,7 @@
 from toad.widgets.presentation_window import protected_presentations
 
 if TYPE_CHECKING:
+    from textual._compositor import SubtreeGeometryPlacement
     from toad.widgets.tool_call import ToolCall
     from toad.widgets.transcript_history import TranscriptPageAdmission
     from toad.transcript_source_preparation import TranscriptSourcePreparation
@@ -145,7 +146,7 @@ def _on_mouse_scroll_right(self, event: events.MouseScrollRight) -> None:
     _restoring = False
     history_anchor: HistoryAnchor | None = None
     history_layout_ready: asyncio.Event | None = None
-    _history_mutation_root: Widget | None = None
+    history_mutation: tuple[Widget, SubtreeGeometryPlacement | None] | None = None
 
     def action_scroll_end(self) -> None:
         self.jump_to_latest()
@@ -181,6 +182,14 @@ def restore_history_layout(self, position: HistoryAnchor) -> bool:
         return self.scroll_y != previous
 
     def finish_history_layout(self) -> None:
+        if self.history_mutating():
+            return
+        if self.history_anchor is not None:
+            # This witness belongs to the actual compensated native layout.
+            # A later source write or scroll retires it through the same epoch.
+            self.history_anchor = replace(
+                self.history_anchor, geometry_revision=self._geometry_revision,
+            )
         if self.history_layout_ready is not None:
             self.history_layout_ready.set()
 
@@ -277,14 +286,8 @@ def history_mutating(self) -> bool:
 
     @property
     def history_mutation_root(self) -> Widget | None:
-        """The subtree owned by the active native publication transaction.
-
-        A direct native window lock still protects the whole window. Scoped
-        history publications name their original mutation owner instead.
-        """
-        if not self.lock.is_locked:
-            return None
-        return self if self._history_mutation_root is None else self._history_mutation_root
+        """The original root whose acquired geometry is held by its writer."""
+        return self.history_mutation[0] if self.history_mutation is not None else None
 
     def visible_history_items(self, items):
         """Borrow this window's clipped cohort from the original native scene."""
@@ -347,21 +350,32 @@ async def preserve_history(self, widget: Widget | None, *, root: Widget | None =
                 # Acquire the native mutation before borrowing an outstanding
                 # anchor. Its owner cannot finish layout while this mutation
                 # holds the tree fence. Release that fence before compensation.
-                previous = self._history_mutation_root
+                screen = self.screen
+                previous = self.history_mutation
                 mutation = self if root is None else root
                 if mutation is not self and self not in mutation.ancestors:
                     raise ValueError("History mutation must belong to its window")
                 if previous is not None:
-                    if previous is mutation or previous in mutation.ancestors:
-                        mutation = previous
-                    elif mutation not in previous.ancestors:
-                        mutation = Widget.get_common_ancestor(previous, mutation)
-                self._history_mutation_root = mutation
+                    previous_root, _source = previous
+                    if previous_root is mutation or previous_root in mutation.ancestors:
+                        mutation = previous_root
+                    elif mutation not in previous_root.ancestors:
+                        mutation = Widget.get_common_ancestor(previous_root, mutation)
+                # Complete source geometry is a writer resource, not the cost
+                # of each visible scroll. Acquire before declaring mutation;
+                # nested expansion lends the previous original held source.
+                self.history_mutation = (
+                    previous if previous is not None and mutation is previous[0]
+                    else (mutation, screen._compositor.acquire_subtree_geometry(mutation))
+                )
                 try:
                     await reader.enter_async_context(self.preserve_reader(widget))
                     yield
                 finally:
-                    self._history_mutation_root = previous
+                    self.history_mutation = previous
+                    # Resume the native owner's retained source requests.
+                    # Releasing a writer does not invent a window-sized layout.
+                    screen.check_idle()
 
     @asynccontextmanager
     async def preserve_reader(self, widget: Widget | None):
@@ -382,23 +396,24 @@ async def preserve_reader(self, widget: Widget | None):
         self.history_anchor = HistoryAnchor.capture(widget, self) if widget is not None else None
         if self.history_anchor is not None and isinstance(screen, WorkspaceScreen):
             screen.viewport_presentation.anchors.add(self)
+        if self.history_anchor is not None:
+            # Layout may complete while source preparation awaits. Retain its
+            # actual completion instead of losing it and forcing another frame.
+            self.history_layout_ready = asyncio.Event()
         geometry = self._geometry_revision
         try:
-            try:
-                yield
-            finally:
-                # The native source owns invalidation. An unchanged page or
-                # already-live body must not manufacture another reflow.
-                if self._geometry_revision != geometry:
-                    self.refresh(layout=True)
+            yield
             if self._geometry_revision == geometry:
                 return
             if (widget is not None and widget.is_attached and self.is_attached
                     and screen.is_current and self.document_viewport.accepts_frame()):
-                # A generic after-refresh callback can run before the pending
-                # mount's layout. Wait for an actual compensated reflow first.
-                self.history_layout_ready = asyncio.Event()
-                await self.history_layout_ready.wait()
+                anchor = self.history_anchor
+                if anchor is not None and anchor.geometry_revision != self._geometry_revision:
+                    # A preceding layout cannot finish a later source write.
+                    # Its original requests remain with Screen until release.
+                    self.history_layout_ready.clear()
+                    screen.check_idle()
+                    await self.history_layout_ready.wait()
         finally:
             if isinstance(screen, WorkspaceScreen):
                 screen.viewport_presentation.anchors.discard(self)
@@ -411,15 +426,18 @@ class HistoryAnchor(WindowRestoration):
     widget: Widget
     scroll_y: float
     scroll_revision: int
+    geometry_revision: int = field(kw_only=True)
     follow_tail: ClassVar[bool]
 
     @classmethod
     def capture(cls, widget: Widget, window: HistoryWindow) -> "HistoryAnchor":
         # Screen coordinates may still describe the frame before a scroll event.
         if window.follows_tail:
-            return TailAnchor(widget, window.scroll_y, window.scroll_revision)
+            return TailAnchor(widget, window.scroll_y, window.scroll_revision,
+                              geometry_revision=window._geometry_revision)
         return RecordAnchor(
             widget, window.scroll_y, window.scroll_revision, cls._offset(widget, window),
+            geometry_revision=window._geometry_revision,
         )
 
     def before_layout(self, window: HistoryWindow) -> HistoryAnchor:
diff --git a/src/toad/widgets/viewport_body.py b/src/toad/widgets/viewport_body.py
index 3877867bf..94cb38352 100644
--- a/src/toad/widgets/viewport_body.py
+++ b/src/toad/widgets/viewport_body.py
@@ -22,6 +22,7 @@
 )
 
 from textual.widget import Widget
+from textual._compositor import SubtreeGeometryPlacement
 from textual.walk import walk_depth_first
 from textual._measurement import INDEPENDENT_HEIGHT, NATIVE_WIDGET_HEIGHT, height_dependency
 from textual.geometry import Size
@@ -1071,7 +1072,9 @@ def visible_bodies(self, windows):
                             yield ancestor, body
                         break
 
-    def mutation_roots(self, windows: tuple[Widget, ...] | None = None) -> tuple[Widget, ...]:
+    def mutation_roots(
+        self, windows: tuple[Widget, ...] | None = None,
+    ) -> dict[Widget, SubtreeGeometryPlacement | None]:
         """Use the acquired frame windows, or acquire them for layout/input."""
         if windows is None:
             windows = tuple(self.frame_windows())
@@ -1079,8 +1082,9 @@ def mutation_roots(self, windows: tuple[Widget, ...] | None = None) -> tuple[Wid
         for anchor in self.anchors:
             if anchor not in displayed and anchor.document_viewport.membership.displayed():
                 displayed[anchor] = None
-        return tuple(root for window in displayed if window.is_attached
-                     if (root := window.history_mutation_root) is not None)
+        return {root: placement for window in displayed if window.is_attached
+                if (mutation := window.history_mutation) is not None
+                for root, placement in (mutation,)}
 
     def prepare(self) -> tuple[Widget, ...]:
         screen = self.screen
diff --git a/tests/history_reflow_travel_pilot.py b/tests/history_reflow_travel_pilot.py
index 56bdfce52..6a185a986 100644
--- a/tests/history_reflow_travel_pilot.py
+++ b/tests/history_reflow_travel_pilot.py
@@ -140,7 +140,7 @@ def observe_anchors():
             with observe_anchors():
                 async with window.preserve_history(None, root=marker):
                     assert window.history_mutation_root is marker
-                    assert app.screen.viewport_presentation.mutation_roots() == (marker,)
+                    assert tuple(app.screen.viewport_presentation.mutation_roots()) == (marker,)
                     async with window.preserve_history(None, root=histories[0]):
                         assert window.history_mutation_root is view.contents
                     assert window.history_mutation_root is marker
diff --git a/tests/retained_fragment_geometry_pilot.py b/tests/retained_fragment_geometry_pilot.py
index 5552a058a..d841cb4c1 100644
--- a/tests/retained_fragment_geometry_pilot.py
+++ b/tests/retained_fragment_geometry_pilot.py
@@ -173,7 +173,7 @@ def trace(frame, event, arg):
             await acquire_capture(body)
             assert await body.retire_body()
             await pilot.pause()
-            retired_scene, _ = compare()
+            retired_scene, _, _placement = compare()
             assert not old_children & retired_scene.keys()
             await body.restore_body()
             await pilot.pause()
@@ -513,7 +513,7 @@ async def second_source():
             receipt['retirement_and_restore_invalidated'] = True
             await history.remove()
             await pilot.pause()
-            final_scene, _ = compare()
+            final_scene, _, _placement = compare()
             assert not set(bodies) & final_scene.keys()
             assert not set(bodies) & scene._subtree_geometry.keys()
             assert app._exception is None
diff --git a/tests/test_worker_preparation_acquisition.py b/tests/test_worker_preparation_acquisition.py
index 3c43791fc..994fe281f 100644
--- a/tests/test_worker_preparation_acquisition.py
+++ b/tests/test_worker_preparation_acquisition.py
@@ -58,6 +58,8 @@ async def pending_publication():
             assert window.history_mutation_root is None
             assert not window.lock.is_locked
             async with window.lock:
+                assert window.history_mutation_root is None
+            async with window.preserve_history(None):
                 assert window.history_mutation_root is window
             assert window.history_mutation_root is None
             assert app._exception is None
diff --git a/tests/viewport_body_lifetime_pilot.py b/tests/viewport_body_lifetime_pilot.py
index 8c7effe08..c3504f61c 100644
--- a/tests/viewport_body_lifetime_pilot.py
+++ b/tests/viewport_body_lifetime_pilot.py
@@ -233,9 +233,14 @@ async def main():
             assert sum(doc.body_dormant for doc in docs) >= 20
             assert all(doc.source == source(i) for i, doc in enumerate(docs))
             from toad.widgets.viewport_body import MeasuredBody
-            cold = [doc for doc in docs if type(doc._body_measurement) is MeasuredBody]
-            assert cold, "Original widget budget never exercised cold eviction"
-            dormant = cold[0]
+            # Widget admission prices native controls; immutable paint has its
+            # own byte admission. This small source need not exhaust that pool.
+            # Exercise cold reconstruction through the same real paint release
+            # used by eviction, without changing either production budget.
+            dormant = next(doc for doc in docs if doc.body_dormant
+                           and not view.window.document_viewport.requires_body(doc))
+            dormant.release_paint()
+            assert type(dormant._body_measurement) is MeasuredBody
             assert not dormant.query(MarkdownParagraph), "Cold body retained its native message pumps"
             before = len(app._registry)
             view.window.release_anchor()

From 9ff0c8d66927dd6703e3cc6951b3fb28db2dc19f Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 18:08:58 -0400
Subject: [PATCH 5/7] Observe completed body publications without occupying
 input pumps

---
 src/toad/widgets/prepared_markdown.py  |  8 ++++----
 src/toad/widgets/transcript_history.py | 10 +++++-----
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/src/toad/widgets/prepared_markdown.py b/src/toad/widgets/prepared_markdown.py
index edee6580c..974f4c284 100644
--- a/src/toad/widgets/prepared_markdown.py
+++ b/src/toad/widgets/prepared_markdown.py
@@ -239,10 +239,10 @@ def reconstructible_children(self) -> tuple[Widget, ...]:
 
     def _initialize_document(self, markdown: str | None) -> AwaitComplete:
         # Mount admits the body; its materialization worker owns prepared
-        # content. Observe the original initialization on this body's pump,
-        # after mount, rather than holding the conversation's mount receipt.
-        # The native implementation retains source consumption and TOC order.
-        self.call_later(super()._initialize_document(markdown))
+        # content. Observe that original receipt after completion; waiting on
+        # this pump would hold wheel delivery behind background preparation.
+        # Native source consumption and TOC completion remain unchanged.
+        super()._initialize_document(markdown).call_when_ready(self)
         return AwaitComplete.nothing()
 
     def retire_body_resources(self) -> None:
diff --git a/src/toad/widgets/transcript_history.py b/src/toad/widgets/transcript_history.py
index 64b443f8a..9d32c9424 100644
--- a/src/toad/widgets/transcript_history.py
+++ b/src/toad/widgets/transcript_history.py
@@ -169,7 +169,7 @@ def set_categories(self, selected: frozenset[type[MessageCategory]]) -> None:
         self.selected_categories = selected
         apply_block_filter(self, selected)
         if self.is_attached and not visible and self.message_category in selected and not self.children:
-            self.publish_body(self.materialize_native_body).call_next(self)
+            self.publish_body(self.materialize_native_body).call_when_ready(self)
 
     def compose(self) -> ComposeResult:
         # A filtered source slot retains identity/coverage, not hidden native
@@ -409,14 +409,14 @@ def _follow_source_tail(self) -> bool:
     async def _report_coverage(self, page: TranscriptPage, fragments: tuple[TranscriptFragment, ...]) -> None:
         if self._source_state.reports_coverage:
             from toad.widgets.conversation import Conversation
-            # Transfer covered native display inside the page admission. Join
-            # stream teardown on the original Conversation pump after these
-            # source/tree locks release, never while mount waits for coverage.
+            # Transfer covered native display inside the page admission. The
+            # original retirement receipt owns teardown; observe its completed
+            # result without holding the Conversation pump or mount admission.
             # Standalone saved viewers have no live transcript to transfer.
             # Resolve custody from native ancestry, not a second owner field.
             for ancestor in self.ancestors:
                 if isinstance(ancestor, Conversation):
-                    ancestor.transcript.covered(TranscriptCoverage(page.events), self).call_next(ancestor)
+                    ancestor.transcript.covered(TranscriptCoverage(page.events), self).call_when_ready(ancestor)
                     break
 
     def publish_committed(self) -> None:

From a54a92a0194db337cc39bb0031a544daac9cc4ea Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 19:06:13 -0400
Subject: [PATCH 6/7] Align document integration with published native
 contracts

---
 src/toad/render_processes.py                 |   9 +-
 src/toad/screens/workspace.py                |   3 +-
 src/toad/widgets/history_anchor.py           |  76 +++---
 src/toad/widgets/viewport_body.py            | 253 ++++++-------------
 tests/history_reflow_travel_pilot.py         |   2 +-
 tests/retained_fragment_geometry_pilot.py    |  66 +----
 tests/retained_tool_text_pilot.py            |   3 +-
 tests/test_worker_preparation_acquisition.py |   8 +-
 tests/viewport_body_lifetime_pilot.py        |  16 +-
 9 files changed, 143 insertions(+), 293 deletions(-)

diff --git a/src/toad/render_processes.py b/src/toad/render_processes.py
index c258b85a0..cff4f063e 100644
--- a/src/toad/render_processes.py
+++ b/src/toad/render_processes.py
@@ -24,15 +24,18 @@
 
 
 def _initialize_worker() -> None:
-    """Acquire typed ingress declarations without borrowing an agent identity.
+    """Acquire ingress and rendering declarations without an agent identity.
 
     These application workers serve both rendering and ACP validation. Finish
-    the required SDK imports during their own startup, before the first ordered
-    notification lends them a task. No notification or rendering job is made.
+    the required SDK and body rendering imports during their own startup,
+    before a notification or recent page lends them a task. Editor and patch
+    implementations remain owned by those tasks, rather than Markdown startup.
+    No notification or rendering job is made.
     """
     os.environ.pop("PI_AGENT_ID", None)
     os.environ.pop("AGENT_COMMS_THREAD", None)
     from toad.acp import sdk_boundary  # noqa: F401 -- declare worker ingress capabilities
+    from toad import render_tasks  # noqa: F401 -- declare worker rendering capabilities
 
 
 class RenderProcessPool(Renderer):
diff --git a/src/toad/screens/workspace.py b/src/toad/screens/workspace.py
index 1877fc451..f371dcdc6 100644
--- a/src/toad/screens/workspace.py
+++ b/src/toad/screens/workspace.py
@@ -20,7 +20,6 @@
 from abc import abstractmethod
 
 if TYPE_CHECKING:
-    from textual._compositor import SubtreeGeometryPlacement
     from toad.app import ToadApp
 
 
@@ -159,7 +158,7 @@ def update_node_styles(self, animate: bool = True) -> None:
             # Record that completed update instead of repeating it next resume.
             self._resume_style = self._style_revision()
 
-    def _layout_mutation_roots(self) -> dict[Widget, SubtreeGeometryPlacement | None]:
+    def _layout_mutation_roots(self) -> tuple[Widget, ...]:
         return self.viewport_presentation.mutation_roots()
 
     def _prepare_compositor_refresh(self) -> tuple[Widget, ...]:
diff --git a/src/toad/widgets/history_anchor.py b/src/toad/widgets/history_anchor.py
index b0e4cff22..cd79de943 100644
--- a/src/toad/widgets/history_anchor.py
+++ b/src/toad/widgets/history_anchor.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field, replace
+from dataclasses import dataclass
 from functools import cached_property
 from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
 import asyncio
@@ -18,7 +18,6 @@
 from toad.widgets.presentation_window import protected_presentations
 
 if TYPE_CHECKING:
-    from textual._compositor import SubtreeGeometryPlacement
     from toad.widgets.tool_call import ToolCall
     from toad.widgets.transcript_history import TranscriptPageAdmission
     from toad.transcript_source_preparation import TranscriptSourcePreparation
@@ -146,7 +145,7 @@ def _on_mouse_scroll_right(self, event: events.MouseScrollRight) -> None:
     _restoring = False
     history_anchor: HistoryAnchor | None = None
     history_layout_ready: asyncio.Event | None = None
-    history_mutation: tuple[Widget, SubtreeGeometryPlacement | None] | None = None
+    _history_mutation_root: Widget | None = None
 
     def action_scroll_end(self) -> None:
         self.jump_to_latest()
@@ -182,14 +181,6 @@ def restore_history_layout(self, position: HistoryAnchor) -> bool:
         return self.scroll_y != previous
 
     def finish_history_layout(self) -> None:
-        if self.history_mutating():
-            return
-        if self.history_anchor is not None:
-            # This witness belongs to the actual compensated native layout.
-            # A later source write or scroll retires it through the same epoch.
-            self.history_anchor = replace(
-                self.history_anchor, geometry_revision=self._geometry_revision,
-            )
         if self.history_layout_ready is not None:
             self.history_layout_ready.set()
 
@@ -286,8 +277,14 @@ def history_mutating(self) -> bool:
 
     @property
     def history_mutation_root(self) -> Widget | None:
-        """The original root whose acquired geometry is held by its writer."""
-        return self.history_mutation[0] if self.history_mutation is not None else None
+        """The subtree owned by the active native publication transaction.
+
+        A direct native window lock still protects the whole window. Scoped
+        history publications name their original mutation owner instead.
+        """
+        if not self.lock.is_locked:
+            return None
+        return self if self._history_mutation_root is None else self._history_mutation_root
 
     def visible_history_items(self, items):
         """Borrow this window's clipped cohort from the original native scene."""
@@ -350,32 +347,21 @@ async def preserve_history(self, widget: Widget | None, *, root: Widget | None =
                 # Acquire the native mutation before borrowing an outstanding
                 # anchor. Its owner cannot finish layout while this mutation
                 # holds the tree fence. Release that fence before compensation.
-                screen = self.screen
-                previous = self.history_mutation
+                previous = self._history_mutation_root
                 mutation = self if root is None else root
                 if mutation is not self and self not in mutation.ancestors:
                     raise ValueError("History mutation must belong to its window")
                 if previous is not None:
-                    previous_root, _source = previous
-                    if previous_root is mutation or previous_root in mutation.ancestors:
-                        mutation = previous_root
-                    elif mutation not in previous_root.ancestors:
-                        mutation = Widget.get_common_ancestor(previous_root, mutation)
-                # Complete source geometry is a writer resource, not the cost
-                # of each visible scroll. Acquire before declaring mutation;
-                # nested expansion lends the previous original held source.
-                self.history_mutation = (
-                    previous if previous is not None and mutation is previous[0]
-                    else (mutation, screen._compositor.acquire_subtree_geometry(mutation))
-                )
+                    if previous is mutation or previous in mutation.ancestors:
+                        mutation = previous
+                    elif mutation not in previous.ancestors:
+                        mutation = Widget.get_common_ancestor(previous, mutation)
+                self._history_mutation_root = mutation
                 try:
                     await reader.enter_async_context(self.preserve_reader(widget))
                     yield
                 finally:
-                    self.history_mutation = previous
-                    # Resume the native owner's retained source requests.
-                    # Releasing a writer does not invent a window-sized layout.
-                    screen.check_idle()
+                    self._history_mutation_root = previous
 
     @asynccontextmanager
     async def preserve_reader(self, widget: Widget | None):
@@ -396,24 +382,23 @@ async def preserve_reader(self, widget: Widget | None):
         self.history_anchor = HistoryAnchor.capture(widget, self) if widget is not None else None
         if self.history_anchor is not None and isinstance(screen, WorkspaceScreen):
             screen.viewport_presentation.anchors.add(self)
-        if self.history_anchor is not None:
-            # Layout may complete while source preparation awaits. Retain its
-            # actual completion instead of losing it and forcing another frame.
-            self.history_layout_ready = asyncio.Event()
         geometry = self._geometry_revision
         try:
-            yield
+            try:
+                yield
+            finally:
+                # The native source owns invalidation. An unchanged page or
+                # already-live body must not manufacture another reflow.
+                if self._geometry_revision != geometry:
+                    self.refresh(layout=True)
             if self._geometry_revision == geometry:
                 return
             if (widget is not None and widget.is_attached and self.is_attached
                     and screen.is_current and self.document_viewport.accepts_frame()):
-                anchor = self.history_anchor
-                if anchor is not None and anchor.geometry_revision != self._geometry_revision:
-                    # A preceding layout cannot finish a later source write.
-                    # Its original requests remain with Screen until release.
-                    self.history_layout_ready.clear()
-                    screen.check_idle()
-                    await self.history_layout_ready.wait()
+                # A generic after-refresh callback can run before the pending
+                # mount's layout. Wait for an actual compensated reflow first.
+                self.history_layout_ready = asyncio.Event()
+                await self.history_layout_ready.wait()
         finally:
             if isinstance(screen, WorkspaceScreen):
                 screen.viewport_presentation.anchors.discard(self)
@@ -426,18 +411,15 @@ class HistoryAnchor(WindowRestoration):
     widget: Widget
     scroll_y: float
     scroll_revision: int
-    geometry_revision: int = field(kw_only=True)
     follow_tail: ClassVar[bool]
 
     @classmethod
     def capture(cls, widget: Widget, window: HistoryWindow) -> "HistoryAnchor":
         # Screen coordinates may still describe the frame before a scroll event.
         if window.follows_tail:
-            return TailAnchor(widget, window.scroll_y, window.scroll_revision,
-                              geometry_revision=window._geometry_revision)
+            return TailAnchor(widget, window.scroll_y, window.scroll_revision)
         return RecordAnchor(
             widget, window.scroll_y, window.scroll_revision, cls._offset(widget, window),
-            geometry_revision=window._geometry_revision,
         )
 
     def before_layout(self, window: HistoryWindow) -> HistoryAnchor:
diff --git a/src/toad/widgets/viewport_body.py b/src/toad/widgets/viewport_body.py
index 94cb38352..22609c5ac 100644
--- a/src/toad/widgets/viewport_body.py
+++ b/src/toad/widgets/viewport_body.py
@@ -6,7 +6,7 @@
 """
 
 from contextlib import AsyncExitStack, ExitStack, asynccontextmanager
-from collections.abc import Awaitable, Callable, Coroutine, Iterator
+from collections.abc import Awaitable, Callable, Coroutine
 from collections import OrderedDict
 from functools import partial
 from weakref import WeakSet, ref
@@ -22,7 +22,6 @@
 )
 
 from textual.widget import Widget
-from textual._compositor import SubtreeGeometryPlacement
 from textual.walk import walk_depth_first
 from textual._measurement import INDEPENDENT_HEIGHT, NATIVE_WIDGET_HEIGHT, height_dependency
 from textual.geometry import Size
@@ -223,12 +222,9 @@ def requires_geometry(self, body) -> bool:
     def geometry_targets(self, body):
         return (body,) if self.requires_geometry(body) else ()
 
-    def capture_requested(self, targets=None):
+    def capture_requested(self):
         return self
 
-    def capture_preparation(self, body, targets):
-        """Source publication does not acquire a retirement geometry demand."""
-
     def required(self):
         return self
 
@@ -298,16 +294,13 @@ class LiveBody(MeasuredBody):
     def released(self):
         return ReleasedBody(self.width, self.rows, self.widgets)
 
-    def capture_requested(self, targets=None):
-        return CapturingBody(self.width, self.rows, self.widgets,
-                             targets=() if targets is None else targets)
+    def capture_requested(self):
+        return CapturingBody(self.width, self.rows, self.widgets)
 
     def prepare_publication(self, body):
         # Borrow the original published scene synchronously, before a writer
         # or its byte measurement can change the native tree.
-        capture = body.capture_native_paint(self)
-        return (capture.prepare(body, self) if capture is not None
-                else MeasuredBody(self.width, self.rows, self.widgets).prepare_publication(body))
+        return body.capture_native_paint(self)
 
     def invalidated(self):
         # The original native tree still carries its pixels. Replace the
@@ -342,16 +335,6 @@ async def materialize(self, body):
     def retire(self, body):
         return body.retire_native_body(self)
 
-    def capture_retirement(self, body, children):
-        # Retirement acquires the current scene after the native frame owner
-        # publishes its geometry. Preceding-source capture above must instead
-        # borrow synchronously before that source's writer changes the tree.
-        capture = body.capture_native_paint(self)
-        if capture is not None:
-            return capture.retire(body, self, children)
-        body._update_body_measurement(self.capture_requested())
-        return BodyMeasurement.retire(self, body)
-
 
 @dataclass(frozen=True)
 class CapturingBody(LiveBody):
@@ -364,22 +347,12 @@ class CapturingBody(LiveBody):
     """
 
     capture_pending = True
-    targets: tuple[Widget, ...] = ()
-
-    def capture_requested(self, targets=None):
-        return self if targets is None or targets == self.targets else replace(self, targets=targets)
 
-    def capture_preparation(self, body, targets):
-        if body._body_measurement is self:
-            body._update_body_measurement(self.capture_requested(targets))
+    def capture_requested(self):
+        return self
 
     def geometry_targets(self, body):
-        return (body, *self.targets)
-
-    def capture_retirement(self, body, children):
-        capture = body.capture_native_paint(self)
-        return (capture.retire(body, self, children) if capture is not None
-                else BodyMeasurement.retire(self, body))
+        return (body, *walk_depth_first(body, PreparedPaintSource, with_root=False))
 
     def required(self):
         return LiveBody(self.width, self.rows, self.widgets)
@@ -616,74 +589,6 @@ async def restore(self, body):
             await body.materialize_body()
 
 
-@dataclass(frozen=True)
-class BodyPaintCapture:
-    """Borrowed native paint and its participants until preparation/pruning.
-
-    Compositor supplies this exact cohort once. The temporary acquisition owns
-    its original source witnesses; the retained body owns only immutable rows,
-    so pruning does not keep the retired child widgets and resources alive.
-    """
-
-    size: Size
-    bands: Iterator[list[Strip]]
-    sources: tuple[tuple[PreparedPaintSource, PreparedRichContent], ...]
-    bodies: tuple[tuple[ViewportBody, BodyMeasurement], ...]
-    style_revision: int
-    geometry_revision: int
-    paint_state: PaintState
-    widgets: int
-
-    def current(self, body):
-        return (self.style_revision == body._subtree_style_revision
-                and self.geometry_revision == body._geometry_revision
-                and all(owner.is_attached and owner.prepared_content is resource
-                        for owner, resource in self.sources)
-                and all(owner.is_attached and owner._body_measurement is resource
-                        and resource.ready(owner) for owner, resource in self.bodies))
-
-    def prepare(self, body, current):
-        # A source writer starts as soon as this call returns. Drain its
-        # preceding pixels now, rather than leaving mutable native readers in
-        # the asynchronous byte-preparation task.
-        content = PreparedRichContent(self.size.width, tuple(
-            strip for band in self.bands for strip in band))
-        return self._prepare(body, current, content)
-
-    async def _prepare(self, body, current, content):
-        size_bytes = await body.app.preparation.run_thread(retained_bytes, content)
-        return RenderedBody(
-            current.width, current.rows, self.widgets, content=content,
-            resource_bytes=size_bytes, style_revision=self.style_revision, paint_state=self.paint_state,
-        )
-
-    async def retire(self, body, current, children):
-        rows = []
-        while True:
-            # Geometry is lent only inside next(). Input, source publication
-            # and ordinary frames use their own scene between these bands.
-            # Reacquire original lifetimes before every native read, including
-            # after the last yield; stale captures never reach pruning.
-            if not body.retirement_current(current) or not self.current(body):
-                return False
-            try:
-                rows.extend(next(self.bands))
-            except StopIteration:
-                break
-            await asyncio.sleep(0)
-        rendered = await self._prepare(body, current, PreparedRichContent(self.size.width, tuple(rows)))
-        if not body.retirement_current(current) or not self.current(body) or not rendered.ready(body):
-            return False
-        async with body.retirement_custody() as can_commit:
-            if (not can_commit or not body.retirement_current(current)
-                    or not self.current(body) or not rendered.ready(body)):
-                return False
-            body._update_body_measurement(rendered)
-            body.retire_body_resources()
-            await body.remove_children(children)
-        return True
-
-
 class MeasuredViewportBody(ViewportBody):
     """The body owns Live, Rendered and Measured behavior, not copied flags."""
 
@@ -767,27 +672,14 @@ def invalidate_body(self):
         self._update_body_measurement(self._body_measurement.invalidated())
 
     def _update_body_measurement(self, measurement):
-        previous = self._body_measurement
-        if measurement is previous:
+        if measurement is self._body_measurement:
             return
-        # State/resource publication is not itself a new layout. Native extent,
-        # descendant participation and geometry demand own that decision. Byte
-        # accounting and writer identity still repaint/request reconciliation
-        # without arranging every unchanged body in the window again.
-        layout = (
-            (measurement.width, measurement.rows) != (previous.width, previous.rows)
-            or measurement.paint_ready(self) != previous.paint_ready(self)
-            or measurement.geometry_targets(self) != previous.geometry_targets(self)
-        )
         self._body_measurement = measurement
         # The resource owns descendant participation and cover selection as
         # well as extent. Publish that change before any native prune awaits;
         # NodeList removal happens later and cannot invalidate it for us.
-        self.refresh(layout=layout)
+        self.refresh(layout=True)
         if self._body_viewport is not None:
-            if measurement.capture_pending:
-                self.screen.frame_presentation.defer(
-                    self._body_viewport.window, self._body_viewport.request)
             self._body_viewport.request()
 
     def notify_style_update(self):
@@ -852,8 +744,8 @@ def retire_body_resources(self):
         pass
 
     def retire_body(self):
-        # Acquire geometry and resource witnesses synchronously. Retirement
-        # paints bands in its original task before byte preparation/pruning.
+        # Acquire pixels synchronously. The viewport can capture its bounded
+        # cohort from one publication before any preparation or pruning awaits.
         return self._body_measurement.retire(self)
 
     @asynccontextmanager
@@ -871,51 +763,65 @@ def retire_native_body(self, current):
         children = self.reconstructible_children()
         if not children:
             return BodyMeasurement.retire(current, self)
-        screen = self.screen
-        if self._after_refresh_pending(screen, screen._layout_mutation_roots()):
+        sources = self.prepared_paint_sources()
+        if not self.prepared_paint_is_current(sources):
             self._update_body_measurement(current.capture_requested())
-            screen.frame_presentation.defer(self._body_viewport.window, self._body_viewport.request)
             return BodyMeasurement.retire(current, self)
-        return current.capture_retirement(self, children)
-
-    def capture_native_paint(self, current):
+        paint = self.capture_native_paint(current, sources=sources)
+        return self.finish_native_retirement(current, children, paint, sources)
+
+    def prepared_paint_sources(self):
+        """Borrow original committed resources, including nested body contents."""
+        return tuple((child, child.prepared_content)
+                     for child in walk_depth_first(self, with_root=True)
+                     if isinstance(child, PreparedPaintSource))
+
+    def prepared_paint_is_current(self, sources):
+        current = self.prepared_paint_sources()
+        return (len(current) == len(sources)
+                and all(owner is original and resource is not None and resource is prepared
+                        for (owner, resource), (original, prepared) in zip(current, sources))
+                and all(child.body_ready for child in walk_depth_first(self, with_root=False)
+                        if isinstance(child, ViewportBody)))
+
+    def capture_native_paint(self, current, *, sources=None):
         """Capture once for retirement and preceding-source publication."""
+        if sources is None:
+            sources = self.prepared_paint_sources()
+        unprepared = MeasuredBody(current.width, current.rows, current.widgets)
         if (not current.ready(self) or not self.is_attached or self.lock.is_locked
                 or any(self in endpoint.walk_ancestors(with_self=True)
-                       for endpoint in self.screen._interaction_widgets())):
-            return None
+                       for endpoint in self.screen._interaction_widgets())
+                or not self.prepared_paint_is_current(sources)):
+            return unprepared.prepare_publication(self)
         compositor = self.screen._compositor
+        style_revision = self._subtree_style_revision
+        paint_state = self._resolved_paint_state()
         captured = tuple(compositor.published_geometry((self,)))
         if not captured:
-            return None
+            return unprepared.prepare_publication(self)
         _body, placement = captured[0]
-        sources = ()
-        bodies = ()
-
-        def admit(participants):
-            nonlocal sources, bodies
-            sources = tuple((owner, owner.prepared_content) for owner in participants
-                            if isinstance(owner, PreparedPaintSource))
-            bodies = tuple((owner, owner._body_measurement) for owner in participants
-                           if owner is not self and isinstance(owner, ViewportBody))
-            if (any(resource is None for _, resource in sources)
-                    or any(not resource.ready(owner) for owner, resource in bodies)):
-                current.capture_preparation(self, tuple(dict.fromkeys(
-                    (*(owner for owner, _ in bodies), *(owner for owner, _ in sources)))))
-                return False
-            return True
-
-        native = compositor.render_subtree_strips(self, placement, admit=admit)
-        if native is None:
-            return None
-        size, bands = native
-        # Witnesses precede the first paint or await. Native bands already
-        # supply styled terminal rows; only detached byte accounting belongs
-        # in preparation workers, not another Rich rendering of those rows.
-        return BodyPaintCapture(
-            size, bands, sources, bodies, self._subtree_style_revision,
-            self._geometry_revision, self._resolved_paint_state(), self.materialized_widget_count,
-        )
+        size, rows = compositor.render_subtree_strips(self, placement)
+        # Arrangement may acquire a different width and start preparation.
+        # Its provisional rows cannot become the source's retained resource.
+        if not self.prepared_paint_is_current(sources):
+            return unprepared.prepare_publication(self)
+        # Native capture already owns final styled terminal rows. Retain those
+        # rows directly; rendering them again in Rich workers duplicates work
+        # and serializes a resource that never leaves this process.
+        content = PreparedRichContent(size.width, tuple(rows))
+        widgets = self.materialized_widget_count
+
+        async def measured():
+            size_bytes = await self.app.preparation.run_thread(retained_bytes, content)
+            if not self.prepared_paint_is_current(sources):
+                return unprepared
+            return RenderedBody(
+                current.width, current.rows, widgets, content=content,
+                resource_bytes=size_bytes, style_revision=style_revision, paint_state=paint_state,
+            )
+
+        return measured()
 
     def retirement_current(self, current):
         """Borrow current viewport demand again after asynchronous capture."""
@@ -937,6 +843,23 @@ async def release_native_body(self, current):
             await self.remove_children(children)
         return True
 
+    async def finish_native_retirement(self, current, children, paint, sources):
+        rendered = await paint
+        # The original state is the captured source/width/style lifetime. An
+        # asynchronous change invalidates it rather than copying a revision.
+        if not self.retirement_current(current):
+            return False
+        if not rendered.ready(self):
+            return False
+        async with self.retirement_custody() as can_commit:
+            if (not can_commit or not self.retirement_current(current) or not rendered.ready(self)
+                    or not self.prepared_paint_is_current(sources)):
+                return False
+            self._update_body_measurement(rendered)
+            self.retire_body_resources()
+            await self.remove_children(children)
+        return True
+
     async def restore_body(self):
         previous = self._body_measurement
         await previous.restore(self)
@@ -1072,9 +995,7 @@ def visible_bodies(self, windows):
                             yield ancestor, body
                         break
 
-    def mutation_roots(
-        self, windows: tuple[Widget, ...] | None = None,
-    ) -> dict[Widget, SubtreeGeometryPlacement | None]:
+    def mutation_roots(self, windows: tuple[Widget, ...] | None = None) -> tuple[Widget, ...]:
         """Use the acquired frame windows, or acquire them for layout/input."""
         if windows is None:
             windows = tuple(self.frame_windows())
@@ -1082,9 +1003,8 @@ def mutation_roots(
         for anchor in self.anchors:
             if anchor not in displayed and anchor.document_viewport.membership.displayed():
                 displayed[anchor] = None
-        return {root: placement for window in displayed if window.is_attached
-                if (mutation := window.history_mutation) is not None
-                for root, placement in (mutation,)}
+        return tuple(root for window in displayed if window.is_attached
+                     if (root := window.history_mutation_root) is not None)
 
     def prepare(self) -> tuple[Widget, ...]:
         screen = self.screen
@@ -1483,11 +1403,6 @@ async def _reconcile(self) -> None:
                         return
                 if self._pending:
                     continue
-                if screen.frame_presentation.awaits_publication(self.window, self.request):
-                    # Foreground restoration remains free to prepare all
-                    # exposed siblings. Retirement resumes from the original
-                    # native callback after their geometry and paint publish.
-                    return
                 retiring = tuple(owner for owner in owners
                                  if owner.is_attached and not owner._closing
                                  and owner not in retained and not owner.body_dormant)
@@ -1500,8 +1415,8 @@ async def _reconcile(self) -> None:
                 retiring = retiring[:self.budget.admission_items]
                 retired_owners = []
                 for first in range(0, len(retiring), self.budget.admission_items):
-                    # Each call acquires the original geometry/resources now;
-                    # tasks paint with lifetime checks between native bands.
+                    # Each call captures its original rows now; tasks only
+                    # prepare detached rows and validate/prune afterward.
                     batch = retiring[first:first + self.budget.admission_items]
                     with ExitStack() as captures:
                         operations = []
diff --git a/tests/history_reflow_travel_pilot.py b/tests/history_reflow_travel_pilot.py
index 6a185a986..56bdfce52 100644
--- a/tests/history_reflow_travel_pilot.py
+++ b/tests/history_reflow_travel_pilot.py
@@ -140,7 +140,7 @@ def observe_anchors():
             with observe_anchors():
                 async with window.preserve_history(None, root=marker):
                     assert window.history_mutation_root is marker
-                    assert tuple(app.screen.viewport_presentation.mutation_roots()) == (marker,)
+                    assert app.screen.viewport_presentation.mutation_roots() == (marker,)
                     async with window.preserve_history(None, root=histories[0]):
                         assert window.history_mutation_root is view.contents
                     assert window.history_mutation_root is marker
diff --git a/tests/retained_fragment_geometry_pilot.py b/tests/retained_fragment_geometry_pilot.py
index d841cb4c1..44ec6dacd 100644
--- a/tests/retained_fragment_geometry_pilot.py
+++ b/tests/retained_fragment_geometry_pilot.py
@@ -16,7 +16,6 @@
 from agent_comms.comms import Comms
 from textual._compositor import Compositor
 from toad.app import ToadApp
-from toad.rich_preparation import PreparedPaintSource
 from toad.widgets.transcript_history import TranscriptHistory
 
 
@@ -64,16 +63,11 @@ async def settle():
             async def acquire_capture(member):
                 """Exercise the body's original complete-row acquisition."""
                 async with asyncio.timeout(10):
-                    while True:
-                        await pilot.pause(.02)
-                        if not member._after_refresh_pending(app.screen, app.screen._layout_mutation_roots()):
-                            capture = member.capture_native_paint(member._body_measurement)
-                            if (capture is not None and capture.current(member)
-                                    and not member._after_refresh_pending(app.screen, app.screen._layout_mutation_roots())):
-                                return
+                    while not member.prepared_paint_is_current(member.prepared_paint_sources()):
                         if member.body_ready and not member.body_capture_pending:
                             assert not await member.retire_body()
                             assert member.body_capture_pending
+                        await pilot.pause(.02)
 
             await settle()
             live_targets = tuple(owner for owner in viewport.body_roots()
@@ -160,20 +154,10 @@ def trace(frame, event, arg):
             await pilot.pause()
             compare()
             receipt['resize_invalidated'] = True
-            # Native exposure owns source custody even in this parked fixture.
-            # The styled fragment is visible here and may not be pruned.
-            assert not await body.retire_body()
-            window.styles.height = 4
-            await pilot.pause()
-            window.release_anchor()
-            window.scroll_to(y=window.max_scroll_y, animate=False, immediate=True)
-            await pilot.pause()
-            assert not viewport.requires_body(body)
             old_children = set(body.walk_children())
-            await acquire_capture(body)
             assert await body.retire_body()
             await pilot.pause()
-            retired_scene, _, _placement = compare()
+            retired_scene, _ = compare()
             assert not old_children & retired_scene.keys()
             await body.restore_body()
             await pilot.pause()
@@ -181,13 +165,9 @@ def trace(frame, event, arg):
             assert body.body_ready and not old_children & set(body.walk_children())
             # The original body family paints full retained rows on reentry.
             # A restoration may not parse/remount at the same width.
-            full_body = "\n\n".join(f"Retained native paragraph {index}." for index in range(40))
-            family = [body, PreparedConversationMarkdown("# Native prepared body\n\n" + full_body),
-                      StreamingMarkdown("# Native streaming body\n\n" + full_body, paginate=False)]
-            from textual.widget import Widget
-            trailing_space = Widget()
-            trailing_space.styles.height = 40
-            await window.mount(*family[1:], trailing_space)
+            family = [body, PreparedConversationMarkdown("# Native prepared body\n\nRetained full lines."),
+                      StreamingMarkdown("# Native streaming body\n\nRetained full lines.", paginate=False)]
+            await window.mount(*family[1:])
             await pilot.pause()
             window.release_anchor()
             window.scroll_to(y=window.max_scroll_y, animate=False, immediate=True)
@@ -207,24 +187,6 @@ def trace(frame, event, arg):
             for member in family:
                 await acquire_capture(member)
             scene.reflow(app.screen, app.size)
-            # A real declaration write may arrive between native bands. The
-            # original capture must discard its partial rows and leave every
-            # native child in custody, rather than publish mixed-style paint.
-            for member in family:
-                children = tuple(member.children)
-                original_color = member.styles.color
-                retirement = asyncio.create_task(member.retire_body())
-                asyncio.get_running_loop().call_soon(setattr, member.styles, 'color', '#bada55')
-                try:
-                    assert not await retirement, type(member).__name__
-                    assert not member.body_dormant and tuple(member.children) == children
-                    assert all(child.is_attached for child in children)
-                finally:
-                    member.styles.color = original_color
-                await pilot.pause()
-                scene.reflow(app.screen, app.size)
-                await acquire_capture(member)
-            receipt['between_band_style_write_retains_all_native_children'] = len(family)
             native_captures = Counter()
 
             def observe_capture(frame, event, arg):
@@ -246,8 +208,8 @@ def observe_capture(frame, event, arg):
                         operations.append(operation)
                 finally:
                     sys.setprofile(None)
-                # Acquire each cohort's original geometry/resources before
-                # band painting or first removal can yield to another task.
+                # Borrow rows before preparation can yield and first removal
+                # can invalidate sibling placement in the original scene.
                 assert native_captures == Counter(bodies=len(family)), native_captures
                 assert all(not member.body_dormant for member in family)
                 assert all(await asyncio.gather(*operations))
@@ -263,11 +225,7 @@ def observe_capture(frame, event, arg):
                 published_scene = scene._full_map, scene._visible_map
                 captured_member, placement = next(scene.published_geometry((member,)))
                 assert captured_member is member
-                native = scene.render_subtree_strips(member, placement, admit=lambda participants: True)
-                assert native is not None
-                _, bands = native
-                for _ in bands:
-                    pass
+                scene.render_subtree_strips(member, placement)
                 assert scene._full_map is published_scene[0]
                 assert scene._visible_map is published_scene[1]
                 if not member.body_dormant:
@@ -444,8 +402,7 @@ async def second_source():
                     member.lock.is_locked, member in viewport.protected(),
                     [(type(source).__name__, source.presentation_ready,
                       source.prepared_content is not None, tuple(source.size))
-                     for source in member.body_geometry_targets()
-                     if isinstance(source, PreparedPaintSource)])
+                     for source, _ in member.prepared_paint_sources()])
                 await pilot.pause()
                 family_receipts[-1]['settled_preceding_writer_released_without_replacing_current_paint'] = True
             receipt['rendered_family_reentry'] = family_receipts
@@ -463,7 +420,6 @@ async def second_source():
             assert not await operation
             assert not streaming.body_dormant
             await streaming.finish_stream()
-            await acquire_capture(streaming)
             assert await streaming.retire_body()
             await pilot.pause()
             receipt['inflight_stream_blocks_captured_retirement_commit'] = True
@@ -513,7 +469,7 @@ async def second_source():
             receipt['retirement_and_restore_invalidated'] = True
             await history.remove()
             await pilot.pause()
-            final_scene, _, _placement = compare()
+            final_scene, _ = compare()
             assert not set(bodies) & final_scene.keys()
             assert not set(bodies) & scene._subtree_geometry.keys()
             assert app._exception is None
diff --git a/tests/retained_tool_text_pilot.py b/tests/retained_tool_text_pilot.py
index f390099a3..350071227 100644
--- a/tests/retained_tool_text_pilot.py
+++ b/tests/retained_tool_text_pilot.py
@@ -97,7 +97,8 @@ async def publication_lifetime(app, pilot, tool):
     assert text in set(app.screen._interaction_widgets())
     assert body in window.document_viewport.protected()
     assert tool in protected_blocks(conversation, (tool,))
-    assert body.capture_native_paint(body._body_measurement) is None
+    measured = await body.capture_native_paint(body._body_measurement)
+    assert not isinstance(measured, RenderedBody)
     await tool.update_tool_call(payload("gesture replacement"))
     assert body.query_one(TextContent) is not text
     assert app.screen._select_state is None and text not in app.screen.selections
diff --git a/tests/test_worker_preparation_acquisition.py b/tests/test_worker_preparation_acquisition.py
index 994fe281f..d485e1cad 100644
--- a/tests/test_worker_preparation_acquisition.py
+++ b/tests/test_worker_preparation_acquisition.py
@@ -58,8 +58,6 @@ async def pending_publication():
             assert window.history_mutation_root is None
             assert not window.lock.is_locked
             async with window.lock:
-                assert window.history_mutation_root is None
-            async with window.preserve_history(None):
                 assert window.history_mutation_root is window
             assert window.history_mutation_root is None
             assert app._exception is None
@@ -664,10 +662,10 @@ async def mounted():
                         "content_lock": document._content_lock.locked(),
                         "stream": str(document._stream),
                         "children": len(document.reconstructible_children()),
-                        "capture_admitted": document.capture_native_paint(document._body_measurement) is not None,
+                        "prepared_current": document.prepared_paint_is_current(document.prepared_paint_sources()),
                         "published_placement": str(tuple(app.screen._compositor.published_geometry((document,)))),
-                        "missing_paint": [str(owner) for owner in document.body_geometry_targets()
-                                          if isinstance(owner, PreparedPaintSource) and owner.prepared_content is None],
+                        "missing_paint": [str(owner) for owner, content in document.prepared_paint_sources()
+                                          if content is None],
                         "nested_not_ready": [str(owner) for owner in walk_depth_first(document, MeasuredViewportBody)
                                              if not owner.body_ready],
                     },
diff --git a/tests/viewport_body_lifetime_pilot.py b/tests/viewport_body_lifetime_pilot.py
index c3504f61c..341e0c680 100644
--- a/tests/viewport_body_lifetime_pilot.py
+++ b/tests/viewport_body_lifetime_pilot.py
@@ -16,7 +16,6 @@
 
 from runtime_fixture import ToadApp
 from toad.widgets.agent_response import AgentResponse
-from toad.rich_preparation import PreparedPaintSource
 
 
 async def settled(view, pilot):
@@ -36,9 +35,7 @@ async def settled(view, pilot):
         if output:
             roots = []
             for owner in manager.body_roots():
-                sources = tuple((child, child.prepared_content)
-                                for child in owner.body_geometry_targets()
-                                if isinstance(child, PreparedPaintSource))
+                sources = owner.prepared_paint_sources()
                 roots.append(dict(
                     measurement=type(owner._body_measurement).__name__,
                     ready=owner.body_ready, required=manager.requires_body(owner),
@@ -167,7 +164,7 @@ async def retirement_visibility(output: Path):
             first = AgentResponse("Original body\n\nActual retained paragraph.")
             await view.contents.mount(first)
             async with asyncio.timeout(15):
-                while not first.body_ready:
+                while not first.body_ready or not first.prepared_paint_is_current(first.prepared_paint_sources()):
                     await pilot.pause(.02)
             # Suspend original housekeeping so this check owns one retirement.
             await manager.suspend_source()
@@ -177,7 +174,7 @@ async def retirement_visibility(output: Path):
             await pilot.pause()
             assert not manager.requires_body(first)
             children = first.reconstructible_children()
-            assert children and first.capture_native_paint(first._body_measurement) is not None
+            assert children and first.prepared_paint_is_current(first.prepared_paint_sources())
             operation = first.retire_body()
             # Actual scrolling exposes the body after synchronous acquisition,
             # before the returned operation measures/commits its captured rows.
@@ -233,10 +230,9 @@ async def main():
             assert sum(doc.body_dormant for doc in docs) >= 20
             assert all(doc.source == source(i) for i, doc in enumerate(docs))
             from toad.widgets.viewport_body import MeasuredBody
-            # Widget admission prices native controls; immutable paint has its
-            # own byte admission. This small source need not exhaust that pool.
-            # Exercise cold reconstruction through the same real paint release
-            # used by eviction, without changing either production budget.
+            # Paint and native controls have separate byte and widget admission.
+            # Exercise cold reconstruction using the original paint release;
+            # this small source does not establish real byte-pressure eviction.
             dormant = next(doc for doc in docs if doc.body_dormant
                            and not view.window.document_viewport.requires_body(doc))
             dormant.release_paint()

From 85e91870f18e8c675a99b31993d941f75ecb4d48 Mon Sep 17 00:00:00 2001
From: Tristan Simas <tristan.simas@mail.mcgill.ca>
Date: Thu, 8 Oct 2026 19:22:36 -0400
Subject: [PATCH 7/7] Carry acquired Markdown syntax through document
 preparation

---
 src/toad/markdown_preparation.py             | 105 +++++++++++++---
 src/toad/render_tasks.py                     |  47 +++++--
 src/toad/widgets/agent_response.py           |   3 +
 src/toad/widgets/incoming_message.py         |   8 +-
 src/toad/widgets/outgoing_message.py         |   6 +-
 src/toad/widgets/prepared_markdown.py        |  47 +++++--
 src/toad/widgets/streaming_markdown.py       |  30 +++--
 src/toad/widgets/transcript_fragments.py     | 124 ++++++++++---------
 src/toad/widgets/transcript_history.py       |  16 ++-
 src/toad/widgets/user_input.py               |   8 +-
 tests/test_worker_preparation_acquisition.py |   7 +-
 11 files changed, 286 insertions(+), 115 deletions(-)

diff --git a/src/toad/markdown_preparation.py b/src/toad/markdown_preparation.py
index ec8061cb5..8325798b1 100644
--- a/src/toad/markdown_preparation.py
+++ b/src/toad/markdown_preparation.py
@@ -1,7 +1,11 @@
 """Process-safe conversation parsing and code-fence highlighting data."""
 
+from copy import deepcopy
+from collections.abc import Iterable, Iterator
 from dataclasses import dataclass, field
 from functools import cached_property
+from hashlib import sha256
+import pickle
 
 from markdown_it.token import Token
 from textual.content import Content
@@ -13,9 +17,82 @@
 
 @dataclass(frozen=True)
 class PreparedMarkdownPart:
-    """One bounded native source part, without an invented transcript role."""
-
-    text: str
+    """Source and syntax acquired through the conversation parser declaration.
+
+    Syntax precedes project-link resolution. Every document acquires its own
+    mutable tokens; neither partitioning nor presentation mutates this source.
+    Custom parser grammars retain their native acquisition instead.
+    """
+
+    text: str = field(compare=False)
+    tokens: tuple[Token, ...] = field(compare=False, repr=False)
+    revision: bytes = field(init=False)
+
+    def __post_init__(self) -> None:
+        # Source acquisition/partitioning runs in preparation. Native refresh
+        # compares this revision without walking an arbitrarily nested tree.
+        object.__setattr__(self, "revision", sha256(pickle.dumps((self.text, self.tokens), protocol=5)).digest())
+
+    @classmethod
+    def capture(cls, text: str) -> "PreparedMarkdownPart":
+        from toad.conversation_markdown import parse_markdown_syntax
+
+        return cls(text, tuple(parse_markdown_syntax(text)))
+
+    def acquire_tokens(self) -> list[Token]:
+        return deepcopy(list(self.tokens))
+
+    def select_blocks(self, text: str, first: int, tokens: slice) -> "PreparedMarkdownPart":
+        """Select complete top-level blocks, retaining their closing tokens."""
+        if text == self.text and first == 0 and tokens.start == 0 and tokens.stop == len(self.tokens):
+            return self
+        selected = deepcopy(list(self.tokens[tokens]))
+        for token in self._descendants(selected):
+            if token.map is not None:
+                token.map = [line - first for line in token.map]
+        return PreparedMarkdownPart(text, tuple(selected))
+
+    def select_table_rows(self, ranges: Iterable[tuple[int, int]]) -> Iterator["PreparedMarkdownPart"]:
+        """Repeat the original parsed header, never parse a synthetic table."""
+        lines = self.text.splitlines(keepends=True)
+        rows = {}
+        body_first = body_stop = 0
+        for index, token in enumerate(self.tokens):
+            if token.type == "tbody_open":
+                body_first = index + 1
+            elif token.type == "tbody_close":
+                body_stop = index
+                break
+            elif body_first and token.type == "tr_open":
+                assert token.map is not None
+                rows[token.map[0]] = index
+        table = self.tokens[0]
+        assert table.type == "table_open" and table.map is not None
+        table_stop = table.map[1]
+        for first, stop in ranges:
+            row_first = rows.get(first, body_stop)
+            row_stop = rows.get(stop, body_stop)
+            tokens = deepcopy(list((*self.tokens[:body_first],
+                                    *self.tokens[row_first:row_stop],
+                                    *self.tokens[body_stop:])))
+            length = 2 + min(stop, table_stop) - first
+            for token in self._descendants(tokens):
+                if token.map is None:
+                    continue
+                if token.type == "table_open":
+                    token.map = [0, length]
+                elif token.type == "tbody_open":
+                    token.map = [2, length]
+                elif token.map[0] >= 2:
+                    token.map = [line - first + 2 for line in token.map]
+            yield PreparedMarkdownPart("".join((*lines[:2], *lines[first:stop])), tuple(tokens))
+
+    @staticmethod
+    def _descendants(tokens):
+        for token in tokens:
+            yield token
+            if token.children is not None:
+                yield from PreparedMarkdownPart._descendants(token.children)
 
     @cached_property
     def retained_bytes(self) -> int:
@@ -33,24 +110,20 @@ class PreparedFence:
 class PreparedMarkdown:
     tokens: list[Token]
     fences: dict[FenceKey, PreparedFence]
-    inlines: dict[int, Content] = field(default_factory=dict)
 
     def acquire_inline_content(self, tokens: list[Token]) -> "PreparedMarkdown":
         """Bind native content to the actual delivered, link-resolved tokens.
 
-        Token identities belong to this one acquisition, not the reusable
-        parser result or another document. Native block construction consumes
-        these same token objects and retains the resulting Content itself.
+        The parser's token metadata owns its acquired content. Native block
+        construction consumes these same tokens, without an identity registry.
         """
-        def descendants(source):
-            for token in source:
-                yield token
-                if token.children is not None:
-                    yield from descendants(token.children)
-
-        inlines = {id(token): MarkdownBlock._token_to_content(token)
-                   for token in descendants(tokens) if token.type == "inline"}
-        return PreparedMarkdown(tokens, self.fences, inlines)
+        for token in PreparedMarkdownPart._descendants(tokens):
+            if token.type == "inline":
+                token.meta[PreparedMarkdown] = MarkdownBlock._token_to_content(token)
+        return PreparedMarkdown(tokens, self.fences)
+
+    def inline_content(self, token: Token) -> Content:
+        return token.meta[PreparedMarkdown]
 
 
 def prepare_tokens(tokens: list[Token], ansi: bool, dark: bool) -> PreparedMarkdown:
diff --git a/src/toad/render_tasks.py b/src/toad/render_tasks.py
index 58ea1a200..fe9b13391 100644
--- a/src/toad/render_tasks.py
+++ b/src/toad/render_tasks.py
@@ -3,20 +3,23 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 from textual.document._document import DocumentBase
 from textual.document._wrapped_document import WrappedDocument
-from textual.widgets import PreparedTextArea
 
 from toad.render_backend import ReusableRenderTask
 
 from toad.markdown_preparation import PreparedMarkdown, PreparedMarkdownPart, prepare_tokens
-from toad.widgets.patch_diff import PreparedPatch, prepare_patch
 from toad.rich_preparation import (
     PreparedRichContent,
     RichPresentation,
     RichSource,
 )
 
+if TYPE_CHECKING:
+    from toad.widgets.patch_diff import PreparedPatch
+    from toad.work_preparation import PreparationRuntime, RenderPreparation, WorkKey
+
 
 @dataclass(frozen=True)
 class ReadOnlyDocumentRenderTask(ReusableRenderTask[WrappedDocument]):
@@ -26,6 +29,8 @@ class ReadOnlyDocumentRenderTask(ReusableRenderTask[WrappedDocument]):
     tab_width: int
 
     def execute(self) -> WrappedDocument:
+        from textual.widgets import PreparedTextArea
+
         return PreparedTextArea.prepare_document(self.source, self.width, self.tab_width)
 
     def accept_result(self, result: object) -> WrappedDocument:
@@ -35,15 +40,19 @@ def accept_result(self, result: object) -> WrappedDocument:
 
 
 @dataclass(frozen=True)
-class PatchRenderTask(ReusableRenderTask[PreparedPatch]):
+class PatchRenderTask(ReusableRenderTask["PreparedPatch"]):
     source: str
     ansi: bool
     dark: bool
 
     def execute(self) -> PreparedPatch:
+        from toad.widgets.patch_diff import prepare_patch
+
         return prepare_patch(self.source, self.ansi, self.dark)
 
     def accept_result(self, result: object) -> PreparedPatch:
+        from toad.widgets.patch_diff import PreparedPatch
+
         if not isinstance(result, PreparedPatch):
             raise TypeError("Patch renderer returned an invalid result")
         return result
@@ -51,16 +60,25 @@ def accept_result(self, result: object) -> PreparedPatch:
 
 @dataclass(frozen=True)
 class MarkdownRenderTask(ReusableRenderTask[PreparedMarkdown]):
-    """Parse and highlight one detached body before its independent delivery."""
+    """Acquire syntax once, then highlight before independent delivery."""
 
-    source: str
+    source: str | PreparedMarkdownPart
     ansi: bool
     dark: bool
 
-    def execute(self) -> PreparedMarkdown:
-        from toad.conversation_markdown import parse_markdown_syntax
+    async def preparation_identity(self, work: RenderPreparation, runtime: PreparationRuntime) -> WorkKey:
+        if isinstance(self.source, PreparedMarkdownPart):
+            from toad.work_preparation import WorkKey
 
-        return prepare_tokens(parse_markdown_syntax(self.source), self.ansi, self.dark)
+            # Acquired syntax owns its revision; accounting and delivery copies
+            # cannot change the original preparation identity.
+            return WorkKey(type(work), (type(self), self.source.revision, self.ansi, self.dark), work.scope)
+        return await super().preparation_identity(work, runtime)
+
+    def execute(self) -> PreparedMarkdown:
+        part = (PreparedMarkdownPart.capture(self.source)
+                if isinstance(self.source, str) else self.source)
+        return prepare_tokens(part.acquire_tokens(), self.ansi, self.dark)
 
     def accept_result(self, result: object) -> PreparedMarkdown:
         if not isinstance(result, PreparedMarkdown):
@@ -72,12 +90,21 @@ def accept_result(self, result: object) -> PreparedMarkdown:
 class MarkdownPartsTask(ReusableRenderTask[tuple[PreparedMarkdownPart, ...]]):
     """Partition one original message with the shared Markdown block budget."""
 
-    source: str
+    source: str | PreparedMarkdownPart
+
+    async def preparation_identity(self, work: RenderPreparation, runtime: PreparationRuntime) -> WorkKey:
+        if isinstance(self.source, PreparedMarkdownPart):
+            from toad.work_preparation import WorkKey
+
+            return WorkKey(type(work), (type(self), self.source.revision), work.scope)
+        return await super().preparation_identity(work, runtime)
 
     def execute(self) -> tuple[PreparedMarkdownPart, ...]:
         from toad.widgets.transcript_fragments import RenderBudget
 
-        parts = tuple(PreparedMarkdownPart(text) for text in RenderBudget().split(self.source))
+        part = (PreparedMarkdownPart.capture(self.source)
+                if isinstance(self.source, str) else self.source)
+        parts = tuple(RenderBudget().split(part))
         for part in parts:
             part.retained_bytes
         return parts
diff --git a/src/toad/widgets/agent_response.py b/src/toad/widgets/agent_response.py
index 5398b742e..6e92fbc8f 100644
--- a/src/toad/widgets/agent_response.py
+++ b/src/toad/widgets/agent_response.py
@@ -8,6 +8,7 @@
 from toad.widgets.message_divider import MessageClock, LiveMessageClock, MessageDivider
 from toad.widgets.route_header import RouteHeader
 from toad import response_delivery
+from toad.markdown_preparation import PreparedMarkdownPart
 from toad.widgets.message_filter import (
     CategorizedBlock,
     MessageCategory,
@@ -32,10 +33,12 @@ def message_category(self) -> type[MessageCategory]:
 
     def __init__(self, markdown: str | None = None, *, delivery: response_delivery.ResponseDelivery = response_delivery.UnroutedResponse(),
                  category: type[MessageCategory] | None = None,
+                 markdown_part: PreparedMarkdownPart | None = None,
                  paginate: bool = True, show_divider: bool = True, clock: MessageClock = LiveMessageClock()) -> None:
         self._message_category = category or delivery.category
         super().__init__(
             markdown,
+            markdown_part=markdown_part,
             paginate=paginate,
             **self.dispatch_sync(delivery, clock, show_divider),
         )
diff --git a/src/toad/widgets/incoming_message.py b/src/toad/widgets/incoming_message.py
index 91c24e504..e5d426c7a 100644
--- a/src/toad/widgets/incoming_message.py
+++ b/src/toad/widgets/incoming_message.py
@@ -12,6 +12,7 @@
 from toad.widgets.committed_presentation import CommitParticipant, SequenceClaim
 from toad.widgets.wire_message_handling import WireMessageHandling
 from toad.widgets.message_notifications import MessageNotifications
+from toad.markdown_preparation import PreparedMarkdownPart
 
 
 
@@ -26,10 +27,12 @@ def message_category(self) -> type[MessageCategory]:
     def commit_claim(self) -> SequenceClaim:
         return SequenceClaim(self.sequence)
 
-    def __init__(self, event: IncomingTranscript, *, show_header: bool = True) -> None:
+    def __init__(self, event: IncomingTranscript, *, show_header: bool = True,
+                 markdown_part: PreparedMarkdownPart | None = None) -> None:
         super().__init__()
         self.event = event
         self.show_header = show_header
+        self.markdown_part = markdown_part
 
     @property
     def message_reference(self):
@@ -46,7 +49,8 @@ def compose(self) -> ComposeResult:
             yield MessageDivider(f"Inbound · @{self.event.route.sender}",
                                  clock=MessageClock.recorded(self.event.timestamp))
             yield RouteHeader(self.event.route, incoming=True)
-        yield AgentResponse(self.event.text, show_divider=False).add_class("routed-body")
+        yield AgentResponse(self.event.text, show_divider=False,
+                            markdown_part=self.markdown_part).add_class("routed-body")
         if self.handling_references:
             yield MessageNotifications()
 
diff --git a/src/toad/widgets/outgoing_message.py b/src/toad/widgets/outgoing_message.py
index 2a99492d3..a8d07af68 100644
--- a/src/toad/widgets/outgoing_message.py
+++ b/src/toad/widgets/outgoing_message.py
@@ -11,15 +11,18 @@
 from toad.widgets.message_filter import CategorizedBlock, OutboundCategory
 from toad.widgets.message_notifications import MessageNotifications
 from toad.widgets.wire_message_handling import WireMessageHandling
+from toad.markdown_preparation import PreparedMarkdownPart
 
 
 class OutgoingMessage(WireMessageHandling, ConversationBlock, CategorizedBlock, VerticalGroup):
     DEFAULT_CLASSES = "block"
 
-    def __init__(self, event: SentTranscript, *, show_header: bool = True):
+    def __init__(self, event: SentTranscript, *, show_header: bool = True,
+                 markdown_part: PreparedMarkdownPart | None = None):
         super().__init__()
         self.event = event
         self.show_header = show_header
+        self.markdown_part = markdown_part
 
     @property
     def message_reference(self):
@@ -31,6 +34,7 @@ def message_category(self):
 
     def compose(self) -> ComposeResult:
         yield AgentResponse(self.event.text,
+                            markdown_part=self.markdown_part,
                             delivery=ResponseDelivery.from_route(self.event.routing.reply),
                             show_divider=self.show_header,
                             clock=MessageClock.recorded(self.event.timestamp))
diff --git a/src/toad/widgets/prepared_markdown.py b/src/toad/widgets/prepared_markdown.py
index 974f4c284..ed918074d 100644
--- a/src/toad/widgets/prepared_markdown.py
+++ b/src/toad/widgets/prepared_markdown.py
@@ -20,10 +20,9 @@
 
 from toad.app import ToadApp
 from toad.conversation_markdown import ConversationCodeFence, ConversationMarkdown, _ThreadLocalPathParser
-from toad.markdown_preparation import PreparedMarkdown
+from toad.markdown_preparation import PreparedMarkdown, PreparedMarkdownPart
 from toad.block_content import MarkdownBlockContent
 from toad.render_tasks import MarkdownRenderTask
-from toad.widgets.transcript_fragments import RenderBudget
 from toad.widgets.viewport_body import MeasuredViewportBody
 from toad.widgets.worker_static import WorkerStatic
 
@@ -227,12 +226,32 @@ def __init__(
         self, markdown: str | None = None, *, name: str | None = None,
         id: str | None = None, classes: str | None = None,
         parser_factory: Callable[[], MarkdownIt] | None = None, open_links: bool = False,
+        markdown_part: PreparedMarkdownPart | None = None,
     ) -> None:
         self._prepared_markdown: PreparedMarkdown | None = None
+        self._markdown_part = markdown_part
         factory = self._make_parser if parser_factory is None else parser_factory
         super().__init__(markdown, name=name, id=id, classes=classes,
                          parser_factory=factory, open_links=open_links)
 
+    @property
+    def partitionable_syntax(self) -> bool:
+        """Only the original conversation factory declares this grammar.
+
+        An overridden factory or arbitrary custom parser has no declaration
+        permitting conversation-syntax partitioning, even if today's output
+        happens to look the same.
+        """
+        return getattr(self._parser_factory, "__func__", None) is ConversationMarkdown._make_parser
+
+    def acquired_source(self, markdown: str) -> str | PreparedMarkdownPart:
+        part = self._markdown_part
+        return part if part is not None and part.text == markdown else markdown
+
+    def update_part(self, part: PreparedMarkdownPart) -> AwaitComplete:
+        self._markdown_part = part
+        return self.update(part.text)
+
     def reconstructible_children(self) -> tuple[Widget, ...]:
         """Native resources rebuilt from this document's original source."""
         return tuple(child for child in self.children if isinstance(child, MarkdownBlock))
@@ -259,9 +278,13 @@ def append(self, markdown: str) -> AwaitComplete:
         return self.publish_body(partial(self._append_body_source, markdown))
 
     def _update_body_source(self, markdown: str) -> AwaitComplete:
+        if self._markdown_part is not None and self._markdown_part.text != markdown:
+            self._markdown_part = None
         return super().update(markdown)
 
     def _append_body_source(self, markdown: str) -> AwaitComplete:
+        if self._markdown_part is not None and self._markdown_part.text != self.source + markdown:
+            self._markdown_part = None
         return super().append(markdown)
 
     def on_unmount(self) -> None:
@@ -273,7 +296,7 @@ async def _parse_tokens(
         # Custom parser factories retain their native semantics. This process
         # function implements the declaration-owned conversation/path parser.
         parent = self.parent
-        if not isinstance(parser, _ThreadLocalPathParser):
+        if type(parser) is not _ThreadLocalPathParser:
             tokens = await super()._parse_tokens(parser, markdown, use_thread=use_thread)
             if tokens is not None:
                 prepared = await asyncio.to_thread(
@@ -284,22 +307,24 @@ async def _parse_tokens(
                 self._prepared_markdown = prepared
             return tokens
         if not isinstance(self.app, ToadApp):
-            tokens = (await asyncio.to_thread(parser.parse, markdown)
-                      if use_thread else parser.parse(markdown))
-            prepared = await asyncio.to_thread(
-                PreparedMarkdown(tokens, {}).acquire_inline_content, tokens,
-            )
+            source = self.acquired_source(markdown)
+            def acquire_local():
+                tokens = (parser.resolve_tokens(source.acquire_tokens())
+                          if isinstance(source, PreparedMarkdownPart) else parser.parse(source))
+                return PreparedMarkdown(tokens, {}).acquire_inline_content(tokens)
+
+            prepared = await asyncio.to_thread(acquire_local)
             if self._closing or self._pruning or self.parent is not parent:
                 return None
             self._prepared_markdown = prepared
-            return tokens
+            return prepared.tokens
 
         # Runway and delivery use the same pure preparation key. Filesystem
         # links are resolved only in each independent delivered token resource,
         # after highlighting; they never become reusable renderer inputs.
         while not self._closing and self.is_attached and not self._pruning:
             theme = (self.app.native_ansi_color, self.app.current_theme.dark)
-            request = self.app.render_processes.submit(MarkdownRenderTask(markdown, *theme))
+            request = self.app.render_processes.submit(MarkdownRenderTask(self.acquired_source(markdown), *theme))
             worker = self.run_worker(request, group="markdown-preparation", exit_on_error=False)
             try:
                 prepared = await worker.wait()
@@ -338,7 +363,7 @@ def _get_token_content(self, token: Token, *, block: MarkdownBlock) -> Content:
             # Its declared answer cannot be replaced by the native pure result.
             return super()._get_token_content(token, block=block)
         assert self._prepared_markdown is not None
-        return self._prepared_markdown.inlines[id(token)]
+        return self._prepared_markdown.inline_content(token)
 
     def get_block_class(self, block_name: str) -> type[MarkdownBlock]:
         if block_name in {"fence", "code_block"}:
diff --git a/src/toad/widgets/streaming_markdown.py b/src/toad/widgets/streaming_markdown.py
index 6f518cbbe..dd8b0aed6 100644
--- a/src/toad/widgets/streaming_markdown.py
+++ b/src/toad/widgets/streaming_markdown.py
@@ -42,7 +42,8 @@ def compose(self) -> ComposeResult:
         yield from self._prefix
 
     def _body(self, fragment: PreparedMarkdownPart) -> PreparedConversationMarkdown:
-        return PreparedConversationMarkdown(fragment.text, classes="-message-fragment")
+        return PreparedConversationMarkdown(fragment.text, markdown_part=fragment,
+                                            classes="-message-fragment")
 
     @property
     def retained_source_bytes(self) -> int:
@@ -60,7 +61,7 @@ async def _prepare_parts(self, parts, current: Callable[[], bool]) -> bool:
             if not current():
                 return False
             await asyncio.gather(*(self.app.render_processes.prepare(MarkdownRenderTask(
-                part.text, self.app.native_ansi_color, self.app.current_theme.dark,
+                part, self.app.native_ansi_color, self.app.current_theme.dark,
             )) for part in parts[first:first + self.batch_size]))
         return current()
 
@@ -127,18 +128,30 @@ async def _publish_content(self, source: str, text: str, append: bool,
         async with self._content_lock:
             if not is_current():
                 return
-            if self.body_dormant:
+            # MaterializingBody is dormant while publishing, even when its
+            # previous native roots remain usable. Append needs reconstruction
+            # only after those roots have actually been retired.
+            if not super().reconstructible_children():
                 self._needs_full_markdown_update = True
-            fragments = (await self.app.render_processes.submit(MarkdownPartsTask(source))
-                         if self._paginate and not RenderBudget().fits(source) else ())
+            if self._markdown_part is not None and self._markdown_part.text != source:
+                self._markdown_part = None
+            fragments = (await self.app.render_processes.submit(MarkdownPartsTask(self.acquired_source(source)))
+                         if self._paginate and self.partitionable_syntax
+                         and not RenderBudget().fits(source) else ())
             if not is_current():
                 return
             if len(fragments) <= 1:
+                if fragments:
+                    self._markdown_part = fragments[0]
                 if self.fragments:
                     await self.remove_children(self.fragment_views)
                     self.fragments, self._fragment_views = (), ()
                     self.start = self.stop = 0
-                if append and self.source + text == source and not self._needs_full_markdown_update:
+                    # Paged views replaced the native roots. Reconstruct the
+                    # body before native append can update its final block.
+                    self._needs_full_markdown_update = True
+                if (append and self.source + text == source
+                        and not self._needs_full_markdown_update):
                     await self._append_body_source(text)
                 else:
                     await self._update_body_source(source)
@@ -161,8 +174,9 @@ async def _publish_content(self, source: str, text: str, append: bool,
             for index, child in previous.items():
                 if not is_current():
                     return
-                if selected.start <= index < selected.stop and child.source != fragments[index].text:
-                    await child.update(fragments[index].text)
+                if (selected.start <= index < selected.stop
+                        and child._markdown_part != fragments[index]):
+                    await child.update_part(fragments[index])
             async with window.preserve_history(None, root=self):
                 if not is_current():
                     return
diff --git a/src/toad/widgets/transcript_fragments.py b/src/toad/widgets/transcript_fragments.py
index a5b9d975b..21718b2f6 100644
--- a/src/toad/widgets/transcript_fragments.py
+++ b/src/toad/widgets/transcript_fragments.py
@@ -1,19 +1,18 @@
 """Render-sized transcript fragments; wire records and cursors remain model-owned."""
 
 import asyncio
-from dataclasses import dataclass, replace
+from dataclasses import dataclass, field, replace
 from functools import cached_property
 from collections.abc import Iterator
 from typing import TYPE_CHECKING
-from threading import local
 
 from agent_comms.mro_dispatch import MroDispatch, handles
 from agent_comms.transcript_events import (
     TranscriptEvent, TextTranscript, ContextTranscript, UserTranscript, MarkdownTranscript, ToolTranscript,
 )
-from markdown_it import MarkdownIt
 
 from toad.render_backend import ReusableRenderTask
+from toad.markdown_preparation import PreparedMarkdownPart
 from toad.acp.status import ToolCallStatus
 from toad.widgets.agent_activity import AgentActivityBoundary
 from toad.widgets.message_filter import event_category, keep_events
@@ -22,19 +21,6 @@
     from toad.render_backend import Renderer
     from toad.tool_output import ToolOutputPart
 
-class _FragmentParserState(local):
-    def __init__(self) -> None:
-        self.parser = MarkdownIt("gfm-like")
-
-
-_parser_state = _FragmentParserState()
-
-
-def _fragment_parser() -> MarkdownIt:
-    """Reuse parser machinery within one execution thread, never concurrently."""
-    return _parser_state.parser
-
-
 async def prepare_transcript_fragments(
     events: tuple[TranscriptEvent, ...], pool: "Renderer | None" = None,
     *, continuation: bool = False,
@@ -63,38 +49,39 @@ class RenderBudget:
     def fits(self, text: str) -> bool:
         return len(text) <= self.characters and len(text.splitlines()) <= self.lines
 
-    def _split_table(self, text: str) -> Iterator[str]:
+    def _split_table(self, part: PreparedMarkdownPart) -> Iterator[PreparedMarkdownPart]:
         """Page a large GFM table by rows, repeating its required header."""
-        lines = text.splitlines(keepends=True)
+        lines = part.text.splitlines(keepends=True)
         if len(lines) <= 2:
-            yield text
+            yield part
             return
         header = "".join(lines[:2])
-        rows: list[str] = []
+        first = 2
         row_characters = 0
         row_limit = max(1, self.lines - 2)
         character_limit = max(self.characters, len(header) + 1)
-        for row in lines[2:]:
-            if rows and (
-                len(rows) >= row_limit
+        table = part.tokens[0]
+        assert table.map is not None
+        ranges = []
+        for index, row in enumerate(lines[2:table.map[1]], 2):
+            if index > first and (
+                index - first >= row_limit
                 or len(header) + row_characters + len(row) > character_limit
             ):
-                yield header + "".join(rows)
-                rows = []
+                ranges.append((first, index))
+                first = index
                 row_characters = 0
-            rows.append(row)
             row_characters += len(row)
-        if rows:
-            yield header + "".join(rows)
+        if first < len(lines):
+            ranges.append((first, len(lines)))
+        yield from part.select_table_rows(ranges)
 
-    def split(self, text: str) -> Iterator[str]:
+    def split(self, source: PreparedMarkdownPart) -> Iterator[PreparedMarkdownPart]:
+        text = source.text
         if not text:
             return
         if self.fits(text):
-            # No block can exceed either budget when the entire document fits.
-            # Return the unchanged source; there is no partition decision to
-            # parse, and normal Markdown rendering still handles its syntax.
-            yield text
+            yield source
             return
         lines = text.splitlines(keepends=True)
         offsets = [0]
@@ -105,43 +92,58 @@ def split(self, text: str) -> Iterator[str]:
         # between top-level Markdown blocks so rendering never invents a line
         # break at an arbitrary provider-chunk or character boundary.
         blocks = {
-            token.map[0]: token.type
-            for token in _fragment_parser().parse(text)
+            token.map[0]: (index, token.type)
+            for index, token in enumerate(source.tokens)
             if token.level == 0 and token.map is not None
         }
         starts = sorted({0, *blocks, len(lines)})
-        parts: list[tuple[str, str | None]] = [
-            (text[offsets[start] : offsets[stop]], blocks.get(start))
+        parts = [
+            (start, stop, blocks.get(start, (0, None))[0],
+             blocks.get(stop, (len(source.tokens), None))[0], blocks.get(start, (0, None))[1])
             for start, stop in zip(starts, starts[1:])
         ]
-        pending = ""
+        pending_first = 0
+        pending_stop = 0
+        pending_token_first = 0
+        pending_token_stop = 0
+        pending_characters = 0
         pending_lines = 0
-        for part, kind in parts:
+        def select(first, stop, token_first, token_stop):
+            return source.select_blocks(text[offsets[first]:offsets[stop]], first,
+                                        slice(token_first, token_stop))
+
+        for start, stop, token_first, token_stop, kind in parts:
+            part = text[offsets[start]:offsets[stop]]
             if kind == "table_open" and (
                 len(part) > self.characters or part.count("\n") > self.lines
             ):
-                if pending:
-                    yield pending
-                    pending = ""
+                if pending_characters:
+                    yield select(pending_first, pending_stop, pending_token_first, pending_token_stop)
+                    pending_characters = 0
                     pending_lines = 0
-                yield from self._split_table(part)
+                yield from self._split_table(select(start, stop, token_first, token_stop))
                 continue
             part_lines = part.count("\n") + int(bool(part) and not part.endswith("\n"))
             exceeds = (
-                pending
+                pending_characters
                 and (
-                    len(pending) + len(part) > self.characters
+                    pending_characters + len(part) > self.characters
                     or pending_lines + part_lines > self.lines
                 )
             )
             if exceeds:
-                yield pending
-                pending = ""
+                yield select(pending_first, pending_stop, pending_token_first, pending_token_stop)
+                pending_characters = 0
                 pending_lines = 0
-            pending += part
+            if not pending_characters:
+                pending_first = start
+                pending_token_first = token_first
+            pending_stop = stop
+            pending_token_stop = token_stop
+            pending_characters += len(part)
             pending_lines += part_lines
-        if pending:
-            yield pending
+        if pending_characters:
+            yield select(pending_first, pending_stop, pending_token_first, pending_token_stop)
 
 
 @dataclass(frozen=True)
@@ -149,11 +151,13 @@ class TranscriptFragment:
     events: tuple[TranscriptEvent, ...]
     continuation: bool = False
     starts_agent_activity: bool = False
+    markdown_part: PreparedMarkdownPart | None = field(default=None, kw_only=True)
 
     def blocks(self, *, fragment: bool = False, show_divider: bool = True):
         from toad.widgets.transcript_history import TranscriptBlockConsumer
 
-        consumer = TranscriptBlockConsumer(fragment=fragment, show_divider=show_divider)
+        consumer = TranscriptBlockConsumer(fragment=fragment, show_divider=show_divider,
+                                           markdown_part=self.markdown_part)
         for event in self.events:
             consumer.dispatch_sync(event)
         return consumer.blocks
@@ -198,10 +202,11 @@ def context(self, event: ContextTranscript):
     @handles(UserTranscript, MarkdownTranscript)
     def text(self, event: TextTranscript):
         starts_activity = self.boundary.observe(event_category(event)) if event.starts_activity else False
+        source = PreparedMarkdownPart.capture(event.text)
         self.fragments.extend(
-            TranscriptFragment((replace(event, text=part),), continuation=self.continuation or index > 0,
-                               starts_agent_activity=starts_activity and index == 0)
-            for index, part in enumerate(self.budget.split(event.text) if self.split_text else (event.text,))
+            TranscriptFragment((replace(event, text=part.text),), continuation=self.continuation or index > 0,
+                               starts_agent_activity=starts_activity and index == 0, markdown_part=part)
+            for index, part in enumerate(self.budget.split(source) if self.split_text else (source,))
         )
 
     @handles(ToolTranscript)
@@ -259,22 +264,23 @@ async def prepare_fragments(self, fragments, keep_going, *, batch_size: int) ->
         for first in range(0, len(fragments), batch_size):
             if not keep_going():
                 return
-            await asyncio.gather(*(self.dispatch(event)
+            await asyncio.gather(*(self.dispatch(event, fragment.markdown_part)
                                    for fragment in fragments[first:first + batch_size]
                                    if self.selected is None or keep_events(fragment.events, self.selected)
                                    for event in fragment.events))
 
     @handles(TranscriptEvent)
-    async def undisclosed(self, event: TranscriptEvent) -> None:
+    async def undisclosed(self, event: TranscriptEvent, part: PreparedMarkdownPart | None) -> None:
         # Metadata and tool disclosure contents retain their existing lazy
         # owners. A viewport prediction does not open those disclosures.
         pass
 
-    @handles(MarkdownTranscript)
-    async def markdown(self, event: MarkdownTranscript) -> None:
+    @handles(UserTranscript, MarkdownTranscript)
+    async def markdown(self, event: TextTranscript, part: PreparedMarkdownPart | None) -> None:
         from toad.render_tasks import MarkdownRenderTask
 
-        await self.renderer.prepare(MarkdownRenderTask(event.text, self.ansi, self.dark))
+        await self.renderer.prepare(MarkdownRenderTask(part if part is not None else event.text,
+                                                      self.ansi, self.dark))
 
 
 @dataclass(frozen=True)
diff --git a/src/toad/widgets/transcript_history.py b/src/toad/widgets/transcript_history.py
index 9d32c9424..16f52783f 100644
--- a/src/toad/widgets/transcript_history.py
+++ b/src/toad/widgets/transcript_history.py
@@ -37,6 +37,7 @@
 )
 from toad.response_delivery import ResponseDelivery
 from toad.widgets.prepared_markdown import PreparedContentRange
+from toad.markdown_preparation import PreparedMarkdownPart
 from toad.widgets.agent_response import AgentResponse
 from toad.widgets.agent_thought import AgentThought
 from toad.widgets.tool_call import ToolCall
@@ -60,10 +61,12 @@ class _PublicationRetired(Exception):
     """Unwind an anchor transaction whose source owner no longer publishes."""
 
 class TranscriptBlockConsumer(MroDispatch):
-    def __init__(self, *, fragment: bool, show_divider: bool):
+    def __init__(self, *, fragment: bool, show_divider: bool,
+                 markdown_part: PreparedMarkdownPart | None = None):
         self.blocks: list[Widget] = []
         self.fragment = fragment
         self.show_divider = show_divider
+        self.markdown_part = markdown_part
 
     @handles(ContextTranscript)
     def context(self, event: ContextTranscript):
@@ -73,30 +76,35 @@ def context(self, event: ContextTranscript):
     @handles(UserTranscript)
     def user(self, event: UserTranscript):
         self.blocks.append(UserInput(event.text, claim=TranscriptInputClaim(event),
+                                     markdown_part=self.markdown_part,
                                      show_divider=self.show_divider,
                                      clock=MessageClock.recorded(event.timestamp)))
 
     @handles(IncomingTranscript)
     def incoming(self, event: IncomingTranscript):
         from toad.widgets.incoming_message import IncomingMessage
-        self.blocks.append(IncomingMessage(event, show_header=self.show_divider))
+        self.blocks.append(IncomingMessage(event, show_header=self.show_divider,
+                                           markdown_part=self.markdown_part))
 
     @handles(SentTranscript)
     def sent(self, event: SentTranscript):
         from toad.widgets.outgoing_message import OutgoingMessage
-        self.blocks.append(OutgoingMessage(event, show_header=self.show_divider))
+        self.blocks.append(OutgoingMessage(event, show_header=self.show_divider,
+                                           markdown_part=self.markdown_part))
 
     @handles(AgentTextTranscript)
     def agent(self, event: AgentTextTranscript):
         self.blocks.append(AgentResponse(
             event.text, delivery=ResponseDelivery.from_route(event.routing.reply if event.routing else None),
             category=event_category(event), paginate=not self.fragment,
+            markdown_part=self.markdown_part,
             show_divider=self.show_divider, clock=MessageClock.recorded(event.timestamp),
         ))
 
     @handles(ThinkingTranscript)
     def thinking(self, event: ThinkingTranscript):
-        self.blocks.append(AgentThought(event.text, paginate=not self.fragment))
+        self.blocks.append(AgentThought(event.text, paginate=not self.fragment,
+                                       markdown_part=self.markdown_part))
 
 def transcript_blocks(events: tuple[TranscriptEvent, ...], *, fragment: bool = False,
                       show_divider: bool = True) -> list[Widget]:
diff --git a/src/toad/widgets/user_input.py b/src/toad/widgets/user_input.py
index 6547f73ec..e1e168210 100644
--- a/src/toad/widgets/user_input.py
+++ b/src/toad/widgets/user_input.py
@@ -4,6 +4,7 @@
 from textual.app import ComposeResult
 from textual import containers
 from toad.widgets.prepared_markdown import PreparedConversationMarkdown
+from toad.markdown_preparation import PreparedMarkdownPart
 
 from toad.widgets.non_selectable_label import NonSelectableLabel
 from toad.widgets.message_divider import MessageDivider, MessageClock, LiveMessageClock
@@ -44,9 +45,12 @@ class UserInput(ConversationBlock, SnapshotPresentation, CategorizedBlock, conta
     def message_category(self) -> type[MessageCategory]:
         return UserCategory
 
-    def __init__(self, content: str, *, claim: CommitClaim = CAPTURED_CLAIM, show_divider: bool = True, clock: MessageClock = LiveMessageClock()) -> None:
+    def __init__(self, content: str, *, claim: CommitClaim = CAPTURED_CLAIM,
+                 markdown_part: PreparedMarkdownPart | None = None,
+                 show_divider: bool = True, clock: MessageClock = LiveMessageClock()) -> None:
         super().__init__()
         self.content = content
+        self.markdown_part = markdown_part
         self.show_divider = show_divider
         self.clock = clock
         self.claim = claim
@@ -60,7 +64,7 @@ def compose(self) -> ComposeResult:
             yield MessageDivider("User", clock=self.clock)
         with containers.HorizontalGroup(classes="user-input-body"):
             yield NonSelectableLabel("❯" if self.show_divider else " ", id="prompt")
-            yield PreparedConversationMarkdown(self.content, id="content")
+            yield PreparedConversationMarkdown(self.content, markdown_part=self.markdown_part, id="content")
 
     def get_clipboard_text(self) -> str | None:
         return self.content
diff --git a/tests/test_worker_preparation_acquisition.py b/tests/test_worker_preparation_acquisition.py
index d485e1cad..bec56c6c5 100644
--- a/tests/test_worker_preparation_acquisition.py
+++ b/tests/test_worker_preparation_acquisition.py
@@ -385,9 +385,12 @@ def input():
                 resources = tuple(markdown._prepared_markdown for markdown in
                                   body.query(PreparedConversationMarkdown)
                                   if markdown._prepared_markdown is not None)
-                assert resources and all(resource.inlines for resource in resources)
+                inline_contents = tuple(resource.inline_content(token)
+                                        for resource in resources for token in resource.tokens
+                                        if token.type == "inline")
+                assert resources and inline_contents
                 assert any(span.style.meta.get("@click", "")
-                           for resource in resources for content in resource.inlines.values()
+                           for content in inline_contents
                            for span in content.spans if not isinstance(span.style, str))
                 assert counts["ui_inline_conversions"] == 0
                 assert counts["worker_inline_conversions"] > 0