Add PreparedTextArea, a read-only TextArea prepared off the UI loop

src/textual/widgets/__init__.py

@@ -46,6 +46,7 @@
     from textual.widgets._tabbed_content import TabbedContent, TabPane
     from textual.widgets._tabs import Tab, Tabs
     from textual.widgets._text_area import TextArea
+    from textual.widgets._prepared_text_area import PreparedTextArea
     from textual.widgets._tooltip import Tooltip
     from textual.widgets._tree import Tree
     from textual.widgets._welcome import Welcome
@@ -90,6 +91,7 @@
     "TabPane",
     "Tabs",
     "TextArea",
+    "PreparedTextArea",
     "Tooltip",
     "Tree",
     "Welcome",

src/textual/widgets/__init__.pyi

@@ -38,6 +38,7 @@ from ._tabbed_content import TabPane as TabPane
 from ._tabs import Tab as Tab
 from ._tabs import Tabs as Tabs
 from ._text_area import TextArea as TextArea
+from ._prepared_text_area import PreparedTextArea as PreparedTextArea
 from ._tooltip import Tooltip as Tooltip
 from ._tree import Tree as Tree
 from ._welcome import Welcome as Welcome

src/textual/widgets/_prepared_text_area.py

@@ -0,0 +1,100 @@
+"""Read-only native documents with detached preparation and live reader intent."""
+from __future__ import annotations
+
+import asyncio
+from functools import partial
+
+from textual.document._document import Document, DocumentBase
+from textual.document._document_navigator import DocumentNavigator
+from textual.document._wrapped_document import WrappedDocument
+from textual.worker import Worker
+from textual.widgets._text_area import TextArea
+
+
+class PreparedTextArea(TextArea):
+    """Keep TextArea selection/navigation while preparing plain text off-loop.
+
+    Editors retain TextArea's synchronous load/edit contract. This read-only
+    surface explicitly awaits prepared loads; resizing coalesces on the native
+    worker lifetime and publishes only a matching document/width/tab view.
+    """
+
+    def __init__(self, text: str = "", **kwargs):
+        super().__init__(text, read_only=True, language=None, **kwargs)
+
+    @staticmethod
+    def prepare_document(source: str | DocumentBase, width: int, tab_width: int) -> WrappedDocument:
+        document = Document(source if isinstance(source, str) else source.text)
+        return WrappedDocument(document, width, tab_width)
+
+    async def _prepare_document(self, source: str | DocumentBase,
+                                width: int, tab_width: int) -> WrappedDocument:
+        """Applications may supply their existing detached process renderer."""
+        return await asyncio.to_thread(self.prepare_document, source, width, tab_width)
+
+    def load_text_prepared(self, text: str) -> Worker[None]:
+        return self.run_worker(partial(self._load_prepared, text), group="prepared-document-load",
+                               exclusive=True)
+
+    async def _load_prepared(self, text: str) -> None:
+        original = self.document
+        # Even an unchanged publication owns cancellation of its predecessor.
+        # Keep equality and document lifetime here, not in individual readers.
+        if original.text == text:
+            return
+        source = text
+        while self.is_attached and not self._pruning and self.document is original:
+            width = self.wrap_width_for(original if isinstance(source, str) else source)
+            tabs = self.indent_width
+            prepared = await self._prepare_document(source, width, tabs)
+            if not self.is_attached or self._pruning or self.document is not original:
+                return
+            if (width, tabs) != (self.wrap_width_for(prepared.document), self.indent_width):
+                source = prepared.document
+                continue
+            if (prepared.document.lines == original.lines and
+                    prepared.document.newline == original.newline):
+                return
+            self.history.clear()
+            self._highlight_query = None
+            self._set_document_view(prepared)
+            self._line_cache.clear()
+            self._refresh_size()
+            self.post_message(self.Changed(self).set_sender(self))
+            self.update_suggestion()
+            return
+
+    def _rewrap_and_refresh_virtual_size(self) -> None:
+        if not self.is_attached:
+            return super()._rewrap_and_refresh_virtual_size()
+        if any(worker.node is self and worker.group == "prepared-document-wrap"
+               and not worker.is_finished for worker in self.workers):
+            return
+        self.run_worker(self._prepare_wrapping, group="prepared-document-wrap")
+
+    async def _prepare_wrapping(self) -> None:
+        while self.is_attached and not self._pruning:
+            document = self.document
+            width, tabs = self.wrap_width, self.indent_width
+            prepared = await self._prepare_document(document, width, tabs)
+            if not self.is_attached or self._pruning:
+                return
+            if (document is not self.document or (width, tabs) !=
+                    (self.wrap_width, self.indent_width)):
+                continue
+            # Programmatic native edits may mutate a read-only document while
+            # preparation is outstanding. Compare its actual lines/newline,
+            # not a copied revision or a separate source signature.
+            if (document.lines != prepared.document.lines or
+                    document.newline != prepared.document.newline):
+                continue
+            reader = self.wrapped_document.offset_to_location(self.scroll_offset)
+            prepared.document = document
+            self.wrapped_document = prepared
+            self.navigator = DocumentNavigator(prepared)
+            self._line_cache.clear()
+            self._refresh_size()
+            offset = prepared.location_to_offset(reader)
+            self.scroll_to(x=offset.x, y=offset.y, animate=False, immediate=True)
+            self.record_cursor_width()
+            return

src/textual/widgets/_text_area.py

@@ -1198,12 +1198,16 @@ def _set_document(self, text: str, language: str | None) -> None:
             # Use a regular plain-text document.
             document = Document(text)
 
-        self.document = document
-        self.wrapped_document = WrappedDocument(document, tab_width=self.indent_width)
+        self._set_document_view(WrappedDocument(document, tab_width=self.indent_width))
+        self._rewrap_and_refresh_virtual_size()
+
+    def _set_document_view(self, wrapped_document: WrappedDocument) -> None:
+        """Admit the document and its matching navigation as one native resource."""
+        self.document = wrapped_document.document
+        self.wrapped_document = wrapped_document
         self.navigator = DocumentNavigator(self.wrapped_document)
         self._build_highlight_map()
         self.move_cursor((0, 0))
-        self._rewrap_and_refresh_virtual_size()
 
     @property
     def _visible_line_indices(self) -> tuple[int, int]:
@@ -1276,10 +1280,14 @@ def wrap_width(self) -> int:
 
         Accounts for gutter, scrollbars, etc.
         """
+        return self.wrap_width_for(self.document)
+
+    def wrap_width_for(self, document: DocumentBase) -> int:
+        """Derive wrapping from this surface and the document being admitted."""
         width, _ = self.scrollable_content_region.size
         cursor_width = 1
         if self.soft_wrap:
-            return max(0, width - self.gutter_width - cursor_width)
+            return max(0, width - self.gutter_width_for(document) - cursor_width)
         return 0
 
     def _rewrap_and_refresh_virtual_size(self) -> None:
@@ -1940,10 +1948,14 @@ def gutter_width(self) -> int:
         Returns:
             The cell-width of the line number column. If `show_line_numbers` is `False` returns 0.
         """
+        return self.gutter_width_for(self.document)
+
+    def gutter_width_for(self, document: DocumentBase) -> int:
+        """The line number declaration belongs to the selected native document."""
         # The longest number in the gutter plus two extra characters: `│ `.
         gutter_margin = 2
         gutter_width = (
-            len(str(self.document.line_count - 1 + self.line_number_start))
+            len(str(document.line_count - 1 + self.line_number_start))
             + gutter_margin
             if self.show_line_numbers
             else 0

tests/text_area/test_prepared_document.py

@@ -0,0 +1,55 @@
+"""Detached read-only work preserves the native document/reader contract."""
+from textual.app import App
+from textual.document._document import Selection
+from textual.widgets import PreparedTextArea
+
+
+async def settle(area, pilot):
+    await pilot.pause()
+    while active := [worker for worker in area.workers
+                     if worker.node is area and not worker.is_finished]:
+        for worker in active:
+            await worker.wait()
+        await pilot.pause()
+
+
+async def test_prepared_load_keeps_selection_and_source_reader_on_resize():
+    app = App()
+    async with app.run_test(size=(75, 18)) as pilot:
+        area = PreparedTextArea(show_line_numbers=True)
+        await app.mount(area)
+        text = "\n".join(f"line {i}\t" + "long plain [text] λ " * 15 for i in range(160))
+        await area.load_text_prepared(text).wait()
+        await settle(area, pilot)
+        assert area.text == text
+        document = area.document
+        area.selection = Selection((80, 2), (81, 10))
+        selected = area.selected_text
+        offset = area.wrapped_document.location_to_offset((60, 0))
+        area.scroll_to(y=offset.y, animate=False, immediate=True)
+        await pilot.pause()
+        reader = area.wrapped_document.offset_to_location(area.scroll_offset)
+        wrapped = area.wrapped_document
+        await area.load_text_prepared(text).wait()
+        assert area.document is document and area.wrapped_document is wrapped
+        await pilot.resize_terminal(44, 18)
+        await settle(area, pilot)
+        assert area.document is document and area.selected_text == selected
+        assert area.wrapped_document.offset_to_location(area.scroll_offset)[0] == reader[0]
+
+
+async def test_new_prepared_source_replaces_only_its_original_native_load():
+    app = App()
+    async with app.run_test() as pilot:
+        area = PreparedTextArea()
+        await app.mount(area)
+        area.load_text_prepared("retired source\n" * 2000)
+        await area.load_text_prepared("current source\n" * 300).wait()
+        await settle(area, pilot)
+        assert area.text == "current source\n" * 300
+        assert area.selection == Selection((0, 0), (0, 0))
+        original = area.document
+        area.load_text_prepared("replaced source\n" * 2000)
+        await area.load_text_prepared(area.text).wait()
+        await settle(area, pilot)
+        assert area.document is original