Keep frame requests made during layout and let busy pumps yield

src/textual/message_pump.py

@@ -808,6 +808,12 @@ async def _process_messages_loop(self) -> None:
                                 break
                     await self._flush_next_callbacks()
 
+            if not self._closed and not self._message_queue.empty():
+                # A ready queue and synchronous handlers need not suspend at
+                # any of the awaits above. Complete this delivery before
+                # handing execution to input, peer pumps and frame timers.
+                await asyncio.sleep(0)
+
     def _can_replace_message(self, message: Message, pending: Message) -> bool:
         """Whether this receiver may consume the pending message instead.
 
@@ -879,8 +885,6 @@ def _get_dispatch_methods(
             method_name: Handler method name.
             message: Message object.
         """
-        from textual.widget import Widget
-
         methods_dispatched: set[Callable] | None = None
         message_types = message.__class__.__mro__
         message_mro: list[type[Message]] | None = None
@@ -914,6 +918,11 @@ def _get_dispatch_methods(
                         else:
                             if not message._sender:
                                 continue
+                            # Widget is required only by selector matching.
+                            # Idle, named and unfiltered decorated delivery
+                            # do not acquire this unrelated type boundary.
+                            from textual.widget import Widget
+
                             for attribute, selector in selectors.items():
                                 node = getattr(message, attribute)
                                 if node is None:

src/textual/screen.py

@@ -1346,12 +1346,16 @@ def _on_timer_update(self) -> None:
                 self._dirty_widgets.clear()
                 self._dirty_widgets.add(self)
                 self._repaint_required = False
-            if self._layout_required:
-                self._refresh_layout(scroll=self._scroll_required)
-                self._layout_required = False
-            elif self._scroll_required:
-                self._refresh_layout(scroll=True)
+            # Consume the acquired intents before layout. Resize watchers and
+            # immediate layout-signal subscribers may request another frame
+            # synchronously; those requests belong to the next admission.
+            layout_required, scroll_required = self._layout_required, self._scroll_required
+            self._layout_required = False
             self._scroll_required = False
+            if layout_required:
+                self._refresh_layout(scroll=scroll_required)
+            elif scroll_required:
+                self._refresh_layout(scroll=True)
 
             if self._dirty_widgets:
                 self._compositor.update_widgets(self._dirty_widgets)

tests/test_refresh_callback_owner.py

@@ -4,10 +4,115 @@
 
 from textual._context import active_message_pump
 from textual.app import App, ComposeResult
+from textual.message import Message
 from textual.screen import Screen
 from textual.widgets import Static
 
 
+async def test_busy_sender_hands_execution_to_peer_after_completed_delivery():
+    class Notice(Message):
+        def __init__(self, sequence):
+            super().__init__()
+            self.sequence = sequence
+
+    delivered = []
+    peer_received = asyncio.Event()
+    finished = asyncio.Event()
+    peer_position = []
+
+    class Sender(Static):
+        def on_notice(self, message):
+            message.stop()
+            delivered.append(message.sequence)
+            if message.sequence == 0:
+                self.app.query_one("#peer").call_later(peer_delivery)
+            if message.sequence == 63:
+                finished.set()
+
+    def peer_delivery():
+        peer_position.append(len(delivered))
+        peer_received.set()
+
+    class BurstApp(App):
+        def compose(self):
+            yield Sender("sender", id="sender")
+            yield Static("peer", id="peer")
+
+    app = BurstApp()
+
+    async def drive(pilot):
+        await pilot.pause()
+        sender = app.query_one("#sender")
+        for sequence in range(64):
+            sender.post_message(Notice(sequence))
+        await asyncio.wait_for(peer_received.wait(), 5)
+        await asyncio.wait_for(finished.wait(), 5)
+        assert delivered == list(range(64))
+        assert 0 < peer_position[0] < 64
+        app.exit()
+
+    await asyncio.wait_for(
+        app.run_async(headless=True, size=(40, 8), auto_pilot=drive), 10
+    )
+    assert app._exception is None
+    assert app._task is None
+
+
+@pytest.mark.parametrize("work", ["layout", "scroll"])
+async def test_frame_preserves_requests_from_its_layout_publication(work):
+    class FrameApp(App):
+        def compose(self):
+            child = Static("native source", id="source")
+            child.styles.width = 10
+            child.styles.height = 40
+            yield child
+
+    app = FrameApp()
+
+    async def drive(pilot):
+        await pilot.pause()
+        screen = app.screen
+        source = app.query_one("#source")
+        received = asyncio.Event()
+
+        def request_next_frame(_screen):
+            screen.screen_layout_refresh_signal.unsubscribe(source)
+            if work == "layout":
+                source.styles.width = 20
+                screen.refresh(layout=True)
+            else:
+                screen.scroll_to(y=2, animate=False, immediate=True, force=True)
+
+        screen.screen_layout_refresh_signal.subscribe(
+            source, request_next_frame, immediate=True
+        )
+        screen.refresh(layout=True)
+        screen._on_timer_update()
+        # The completed geometry is still the first frame. Its subscriber
+        # requested independent work through the original native owner.
+        if work == "layout":
+            assert source.region.width == 10
+            assert screen._layout_required
+        else:
+            assert screen.scroll_y == 2
+            assert screen._scroll_required
+        assert screen._refresh_pending
+        screen.call_after_refresh(received.set)
+        await asyncio.wait_for(received.wait(), 5)
+        if work == "layout":
+            assert source.region.width == 20
+        else:
+            assert source.region.y == -2
+        assert not screen._refresh_pending
+        app.exit()
+
+    await asyncio.wait_for(
+        app.run_async(headless=True, size=(40, 8), auto_pilot=drive), 10
+    )
+    assert app._exception is None
+    assert app._task is None
+
+
 @pytest.mark.parametrize("sender_kind", ["app", "widget"])
 async def test_async_sender_callback_does_not_borrow_screen_task(sender_kind):
     """A suspended sender cannot own another sender's paint or callbacks."""