The pilot settles on owed work, not on a CPU-time guess

src/textual/app.py

@@ -88,7 +88,6 @@
     _make_path_object_relative,
 )
 from textual._types import AnimationLevel
-from textual._wait import wait_for_idle
 from textual.actions import ActionParseResult, SkipAction
 from textual.ansi_palette import AnsiPalette, MappedAnsiPalette
 from textual.await_complete import AwaitComplete
@@ -243,6 +242,13 @@ class InvalidThemeError(Exception):
     """Raised when an invalid theme is set."""
 
 
+class WaitForScreenTimeout(Exception):
+    """Exception raised if messages aren't being processed quickly enough.
+
+    If this occurs, the most likely explanation is some kind of deadlock in the app code.
+    """
+
+
 ReturnType = TypeVar("ReturnType")
 CallThreadReturnType = TypeVar("CallThreadReturnType")
 
@@ -2137,13 +2143,110 @@ def get_key_display(self, binding: Binding) -> str:
         key_tokens = modifiers + [key]
         return "+".join(key_tokens)
 
+    SETTLE_UPDATES: ClassVar[int] = 2
+    """Screen updates one settle runs. A layout's own consequences (the Resize it sends
+    re-measuring content) land in the second; a widget that asks for a layout on every
+    layout never stops asking, so the settle stops there."""
+
+    async def _settle(self, timeout: float = 30.0) -> bool:
+        """Handle what is posted to the app and its screen, and whatever that handling posts,
+        until no pump there owes work.
+
+        Two kinds of work are owed. Messages: each round waits behind the work every owing
+        pump has queued, and work it posts makes its target owe again. The update those
+        messages asked for: layout and painting run on the screen's update timer, not in a
+        queue, so once the queues drain the settle runs that update itself, then looks at
+        the queues again. A key's whole cascade (the app forwarding it, the focused widget
+        handling it, the layout and paint it asks for) is done when an update leaves
+        nothing owed. A running animation is not owed: it may run for a minute, and callers
+        that need it finished ask the animator (Pilot.wait_for_animation).
+
+        Args:
+            timeout: Seconds one round may take.
+
+        Returns:
+            `True` once nothing is owed; `False` if an exception stopped the app first.
+
+        Raises:
+            WaitForScreenTimeout: If one round did not finish within the timeout.
+        """
+        updates = 0
+        while True:
+            # Input a driver delivered is posted from a loop callback (call_soon_threadsafe):
+            # one turn of the loop puts everything already handed to it on its queues.
+            await asyncio.sleep(0)
+            try:
+                screen = self.screen
+            except Exception:
+                return False
+            pumps = (self, *screen.walk_children(with_self=True))
+            owed = await self._wait_behind(
+                [pump.call_after_pending for pump in pumps], timeout
+            )
+            if self._exception_event.is_set() or self._exit:
+                return False
+            if owed:
+                continue
+            if self._resize_event is not None:
+                # A terminal resize waits on a debounce timer; deliver it to the screens.
+                self._check_resize()
+                continue
+            if updates == self.SETTLE_UPDATES:
+                return True
+            await screen._update_now()
+            updates += 1
+
+    async def _wait_behind(
+        self, schedulers: list[Callable[[Callable[[], object]], bool]], timeout: float
+    ) -> int:
+        """Schedule one callback with each scheduler and wait until every scheduled one ran,
+        or an exception stopped the app.
+
+        Returns:
+            How many callbacks were scheduled; a scheduler answering False owed no work.
+
+        Raises:
+            WaitForScreenTimeout: If they did not all run within the timeout.
+        """
+        count = 0
+        handled_all = asyncio.Event()
+
+        def handled() -> None:
+            nonlocal count
+            count -= 1
+            if count == 0:
+                handled_all.set()
+
+        scheduled = sum(schedule(handled) for schedule in schedulers)
+        count += scheduled
+        if not scheduled:
+            return 0
+        waits = [
+            create_task(handled_all.wait()),
+            create_task(self._exception_event.wait()),
+        ]
+        # The app's own loop ending (an exit, a crash) means nothing owed will ever run.
+        loop_ended = [] if self._task is None else [self._task]
+        done, _ = await asyncio.wait(
+            [*waits, *loop_ended], timeout=timeout, return_when=asyncio.FIRST_COMPLETED
+        )
+        for task in waits:
+            task.cancel()
+        if not done:
+            raise WaitForScreenTimeout(
+                "Timed out while waiting for widgets to process pending messages."
+            )
+        return scheduled
+
     async def _press_keys(self, keys: Iterable[str]) -> None:
         """A task to send key events."""
         import unicodedata
 
         app = self
-        driver = app._driver
-        assert driver is not None
+        # A typed key arrives after the app handled what it was given before it (a screen
+        # being pushed, a widget mounting): the first key lands on that settled app.
+        if not await app._settle():
+            return
         for key in keys:
             if key.startswith("wait:"):
                 _, wait_ms = key.split(":")
@@ -2160,10 +2263,14 @@ async def _press_keys(self, keys: Iterable[str]) -> None:
                     char = key if len(key) == 1 else None
                 key_event = events.Key(key, char)
                 key_event.set_sender(app)
-                driver.send_message(key_event)
-                await wait_for_idle(0)
+                # This runs on the app's loop: post now, so the settle below sees it owed.
+                # The driver's thread-safe ingress would land it one loop turn later.
+                app.post_message(key_event)
+                if not await app._settle():
+                    # The key stopped the app: the rest have nothing to land on.
+                    return
                 await app._animator.wait_until_complete()
-                await wait_for_idle(0)
+                await app._settle()
 
     def _flush(self, stderr: bool = False) -> None:
         """Called when stdout or stderr is flushed.
@@ -2303,7 +2410,7 @@ async def run_app(app: App[ReturnType]) -> None:
             # Context manager returns pilot object to manipulate the app
             try:
                 pilot = Pilot(app)
-                await pilot._wait_for_screen()
+                await app._settle()
                 yield pilot
             finally:
                 await asyncio.sleep(0)

src/textual/css/styles.py

@@ -507,11 +507,13 @@ class StylesBase(BoxGeometry, PaintLayer, TextFormat):
     constrain_y: StringEnumProperty[Constrain] = StringEnumProperty(
         VALID_CONSTRAIN, "none"
     )
+    # Wrapping and overflow decide how many lines the content measures, so an auto-sized
+    # widget lays out again, like line_pad below.
     text_wrap: StringEnumProperty[TextWrap] = StringEnumProperty(
-        VALID_TEXT_WRAP, TextFormat().line_wrap
+        VALID_TEXT_WRAP, TextFormat().line_wrap, layout=True
     )
     text_overflow: StringEnumProperty[TextOverflow] = StringEnumProperty(
-        VALID_TEXT_OVERFLOW, TextFormat().line_overflow
+        VALID_TEXT_OVERFLOW, TextFormat().line_overflow, layout=True
     )
     expand: StringEnumProperty[Expand] = StringEnumProperty(VALID_EXPAND, "greedy")
     line_pad = IntegerProperty(default=TextFormat().line_pad, layout=True)

src/textual/pilot.py

@@ -13,7 +13,6 @@
 
 import rich.repr
 
-from textual._wait import wait_for_idle
 from textual.app import App, ReturnType
 from textual.drivers.headless_driver import HeadlessDriver
 from textual.events import Click, MouseDown, MouseEvent, MouseMove, MouseUp, Resize
@@ -51,13 +50,6 @@ class OutOfBounds(Exception):
     """Raised when the pilot mouse target is outside of the (visible) screen."""
 
 
-class WaitForScreenTimeout(Exception):
-    """Exception raised if messages aren't being processed quickly enough.
-
-    If this occurs, the most likely explanation is some kind of deadlock in the app code.
-    """
-
-
 @rich.repr.auto(angular=True)
 class Pilot(Generic[ReturnType]):
     """Pilot object to drive an app."""
@@ -81,7 +73,7 @@ async def press(self, *keys: str) -> None:
         """
         if keys:
             await self._app._press_keys(keys)
-            await self._wait_for_screen()
+            await self._app._settle()
 
     async def resize_terminal(self, width: int, height: int) -> None:
         """Resize the terminal to the given dimensions.
@@ -461,101 +453,33 @@ async def _post_mouse_events(
             landed is not None and landed.widget is target_widget
         )
 
-    async def _wait_for_screen(self, timeout: float = 30.0) -> bool:
-        """Wait for the current screen and its children to have processed all pending events.
-
-        Args:
-            timeout: A timeout in seconds to wait.
-
-        Returns:
-            `True` if all events were processed. `False` if an exception occurred,
-            meaning that not all events could be processed.
-
-        Raises:
-            WaitForScreenTimeout: If the screen and its children didn't finish processing within the timeout.
-        """
-        try:
-            screen = self.app.screen
-        except Exception:
-            return False
-        children = [self.app, *screen.walk_children(with_self=True)]
-        count = 0
-        count_zero_event = asyncio.Event()
-
-        def decrement_counter() -> None:
-            """Decrement internal counter, and set an event if it reaches zero."""
-            nonlocal count
-            count -= 1
-            if count == 0:
-                # When count is zero, all messages queued at the start of the method have been processed
-                count_zero_event.set()
-
-        # Increase the count for every successful call_later
-        for child in children:
-            if child.call_after_pending(decrement_counter):
-                count += 1
-
-        if count:
-            # Wait for the count to return to zero, or a timeout, or an exception
-            wait_for = [
-                asyncio.create_task(count_zero_event.wait()),
-                asyncio.create_task(self.app._exception_event.wait()),
-            ]
-            _, pending = await asyncio.wait(
-                wait_for,
-                timeout=timeout,
-                return_when=asyncio.FIRST_COMPLETED,
-            )
-
-            for task in pending:
-                task.cancel()
-
-            timed_out = len(wait_for) == len(pending)
-            if timed_out:
-                raise WaitForScreenTimeout(
-                    "Timed out while waiting for widgets to process pending messages."
-                )
-
-            # We've either timed out, encountered an exception, or we've finished
-            # decrementing all the counters (all events processed in children).
-            if count > 0:
-                return False
-
-        return True
-
     async def pause(self, delay: float | None = None) -> None:
         """Insert a pause.
 
         Args:
-            delay: Seconds to pause, or None to wait for cpu idle.
+            delay: Seconds to pause after the app settles, or None for no extra time.
         """
-        # These sleep zeros, are to force asyncio to give up a time-slice.
-        await self._wait_for_screen()
-        if delay is None:
-            await wait_for_idle(0)
-        else:
+        await self._app._settle()
+        if delay is not None:
             await asyncio.sleep(delay)
-        self.app.screen._on_timer_update()
+            await self._app._settle()
 
     async def wait_for_animation(self) -> None:
         """Wait for any current animation to complete."""
         await self._app.animator.wait_for_idle()
-        self.app.screen._on_timer_update()
+        await self._app._settle()
 
     async def wait_for_scheduled_animations(self) -> None:
         """Wait for any current and scheduled animations to complete."""
-        await self._wait_for_screen()
+        await self._app._settle()
         await self._app.animator.wait_until_complete()
-        await self._wait_for_screen()
-        await wait_for_idle()
-        self.app.screen._on_timer_update()
+        await self._app._settle()
 
     async def exit(self, result: ReturnType) -> None:
         """Exit the app with the given result.
 
         Args:
             result: The app result returned by `run` or `run_async`.
         """
-        await self._wait_for_screen()
-        await wait_for_idle()
+        await self._app._settle()
         self.app.exit(result)

src/textual/screen.py

@@ -1528,6 +1528,17 @@ def _layout_geometry_targets(self) -> Iterable[Widget]:
         """
         return ()
 
+    async def _update_now(self) -> None:
+        """Run the update the timer would, and the rest of a layout it splits into steps.
+
+        The update pauses the timer. Work requested while it ran is only seen by the
+        screen's idle check, which resumes the timer for it, so the screen is prompted.
+        """
+        self._on_timer_update()
+        if (remaining := self._stepped_layout_task) is not None:
+            await remaining
+        self.check_idle()
+
     def _refresh_layout(self, size: Size | None = None, scroll: bool = False) -> None:
         """Refresh the layout (can change size and positions of widgets)."""
         self._stop_stepped_layout()