Cache message class ancestry and handler declarations for dispatch

src/textual/message.py

@@ -45,6 +45,7 @@ class Message:
     namespace: ClassVar[str] = ""  # Namespace to disambiguate messages
     handler_name: ClassVar[str]
     """Name of the default message handler."""
+    _dispatch_types = None
 
     def __init__(self) -> None:
         self.__post_init__()
@@ -87,6 +88,21 @@ def __init_subclass__(
             namespace = qualname.rsplit(".", 2)[-2:]
             name = "_".join(camel_to_snake(part) for part in namespace)
         cls.handler_name = f"on_{name}"
+        mro = cls.__mro__
+        cls._dispatch_types = mro, tuple(
+            ancestor for ancestor in mro if issubclass(ancestor, Message)
+        )
+
+    @classmethod
+    def _get_dispatch_types(cls):
+        """Supply the actual message ancestors from their declaration source."""
+        mro = cls.__mro__
+        source = cls._dispatch_types
+        if source is None or source[0] is not mro:
+            source = cls._dispatch_types = mro, tuple(
+                ancestor for ancestor in mro if issubclass(ancestor, Message)
+            )
+        return source[1]
 
     @property
     def control(self) -> DOMNode | None:

src/textual/message_pump.py

@@ -117,8 +117,26 @@ def __new__(
                     )
 
         class_obj = super().__new__(cls, name, bases, class_dict, **kwargs)
+        mro = class_obj.__mro__
+        class_obj._dispatch_declarations = (
+            mro, tuple((owner, owner.__dict__) for owner in mro)
+        )
         return class_obj
 
+    def _get_dispatch_declarations(cls):
+        """Supply C3 declarations without copying their live dictionaries.
+
+        A bases change, including on a plain mixin, replaces Python's MRO.
+        Method and decorated-table replacements remain visible through the
+        original mapping views, even between yields of one delivery.
+        """
+        mro = cls.__mro__
+        source, declarations = cls._dispatch_declarations
+        if source is not mro:
+            declarations = tuple((owner, owner.__dict__) for owner in mro)
+            cls._dispatch_declarations = mro, declarations
+        return declarations
+
 
 class MessagePump(metaclass=_MessagePumpMeta):
     """Base class which supplies a message pump."""
@@ -845,24 +863,21 @@ def _get_dispatch_methods(
         from textual.widget import Widget
 
         methods_dispatched: set[Callable] | None = None
-        message_types = message.__class__.__mro__
-        message_mro: list[type[Message]] | None = None
+        # Capture both original linearizations before the first yield. A
+        # handler may replace bases while this delivery is suspended; its
+        # successor delivery acquires the changed declaration source.
+        message_mro = message.__class__._get_dispatch_types()
         private_method_name = f"_{method_name}"
-        for cls in self.__class__.__mro__:
+        for cls, declarations in self.__class__._get_dispatch_declarations():
             if message._no_default_action:
                 break
-            declarations = cls.__dict__
             # Try decorated handlers first
             decorated_handlers = cast(
                 "dict[type[Message], list[tuple[Callable, dict[str, tuple[SelectorSet, ...]]]]] | None",
                 declarations.get("_decorated_handlers"),
             )
 
             if decorated_handlers:
-                if message_mro is None:
-                    message_mro = [
-                        _type for _type in message_types if issubclass(_type, Message)
-                    ]
                 for message_class in message_mro:
                     handlers = decorated_handlers.get(message_class, ())
 

tests/test_on.py

@@ -13,6 +13,91 @@
 from textual.widgets import Button, TabbedContent, TabPane
 
 
+async def test_dispatch_retains_acquired_ancestry_and_reacquires_changed_bases():
+    class FirstNotice(Message):
+        pass
+
+    class SecondNotice(Message):
+        pass
+
+    class Notice(FirstNotice, namespace="changing", bubble=False):
+        pass
+
+    class FirstMixin:
+        def on_changing_notice(self):
+            self.received.append("first mixin")
+
+    class SecondMixin:
+        def on_changing_notice(self):
+            self.received.append("second mixin")
+
+    class Pivot(FirstMixin):
+        pass
+
+    class BaseApp(Pivot, App):
+        @on(FirstNotice)
+        def first_notice(self):
+            self.received.append("first notice")
+
+        @on(SecondNotice)
+        def second_notice(self):
+            self.received.append("second notice")
+
+    class ChangingApp(BaseApp):
+        def __init__(self):
+            super().__init__()
+            self.received = []
+
+        def on_changing_notice(self):
+            self.received.append("head")
+            # This plain mixin mutation does not pass through MessagePump's
+            # metaclass. Both original linearizations must remain acquired for
+            # this delivery, while the next delivery sees Python's new MROs.
+            Pivot.__bases__ = (SecondMixin,)
+            Notice.__bases__ = (SecondNotice,)
+
+    app = ChangingApp()
+    try:
+        async with app.run_test() as pilot:
+            app.post_message(Notice())
+            await pilot.pause()
+            assert app.received == ["head", "first notice", "first mixin"]
+            app.received.clear()
+            app.post_message(Notice())
+            await pilot.pause()
+            assert app.received == ["head", "second notice", "second mixin"]
+    finally:
+        Pivot.__bases__ = (FirstMixin,)
+        Notice.__bases__ = (FirstNotice,)
+
+
+async def test_dispatch_reads_replaced_decorated_table_between_yields():
+    class Notice(Message, namespace="table", bubble=False):
+        pass
+
+    class BaseApp(App):
+        @on(Notice)
+        def original(self):
+            self.received.append("original")
+
+    def replacement(instance, message):
+        instance.received.append("replacement")
+
+    class ChangingApp(BaseApp):
+        def __init__(self):
+            super().__init__()
+            self.received = []
+
+        def on_table_notice(self):
+            BaseApp._decorated_handlers = {Notice: [(replacement, {})]}
+
+    app = ChangingApp()
+    async with app.run_test() as pilot:
+        app.post_message(Notice())
+        await pilot.pause()
+        assert app.received == ["replacement"]
+
+
 async def test_live_named_replacement_during_decorated_dispatch():
     class Notice(Message, namespace="lookup", bubble=False):
         pass