diff --git a/CHANGELOG.md b/CHANGELOG.md
index ef62324f2..9d7ec18b5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,13 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## Unreleased
+
+### Changed
+
+- Reactive callbacks are now scheduled on the message pump of the reactable that is watching instead of the owner of reactive attribute https://github.com/Textualize/textual/pull/3065
+- Callbacks scheduled with `call_next` will now have the same prevented messages as when the callback was scheduled https://github.com/Textualize/textual/pull/3065
+
 ## [0.35.0]
 
 ### Added
@@ -69,6 +76,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - DescendantBlur and DescendantFocus can now be used with @on decorator
 
+
 ## [0.32.0] - 2023-08-03
 
 ### Added
diff --git a/src/textual/message.py b/src/textual/message.py
index d80d512e8..931c5aa21 100644
--- a/src/textual/message.py
+++ b/src/textual/message.py
@@ -11,7 +11,6 @@ import rich.repr
 
 from . import _time
 from ._context import active_message_pump
-from ._types import MessageTarget
 from .case import camel_to_snake
 
 if TYPE_CHECKING:
diff --git a/src/textual/message_pump.py b/src/textual/message_pump.py
index affdf0843..7ed468dca 100644
--- a/src/textual/message_pump.py
+++ b/src/textual/message_pump.py
@@ -118,7 +118,7 @@ class MessagePump(metaclass=_MessagePumpMeta):
         self._last_idle: float = time()
         self._max_idle: float | None = None
         self._mounted_event = asyncio.Event()
-        self._next_callbacks: list[CallbackType] = []
+        self._next_callbacks: list[events.Callback] = []
         self._thread_id: int = threading.get_ident()
 
     @property
@@ -417,7 +417,9 @@ class MessagePump(metaclass=_MessagePumpMeta):
             *args: Positional arguments to pass to the callable.
             **kwargs: Keyword arguments to pass to the callable.
         """
-        self._next_callbacks.append(partial(callback, *args, **kwargs))
+        callback_message = events.Callback(callback=partial(callback, *args, **kwargs))
+        callback_message._prevent.update(self._get_prevented_messages())
+        self._next_callbacks.append(callback_message)
         self.check_idle()
 
     def _on_invoke_later(self, message: messages.InvokeLater) -> None:
@@ -562,7 +564,7 @@ class MessagePump(metaclass=_MessagePumpMeta):
         self._next_callbacks.clear()
         for callback in callbacks:
             try:
-                await invoke(callback)
+                await self._dispatch_message(callback)
             except Exception as error:
                 self.app._handle_exception(error)
                 break
diff --git a/src/textual/reactive.py b/src/textual/reactive.py
index 328a45832..d361bd004 100644
--- a/src/textual/reactive.py
+++ b/src/textual/reactive.py
@@ -220,11 +220,15 @@ class Reactive(Generic[ReactiveType]):
             obj.post_message(events.Callback(callback=partial(Reactive._compute, obj)))
 
         def invoke_watcher(
-            watch_function: Callable, old_value: object, value: object
+            watcher_object: Reactable,
+            watch_function: Callable,
+            old_value: object,
+            value: object,
         ) -> None:
             """Invoke a watch function.
 
             Args:
+                watcher_object: The object watching for the changes.
                 watch_function: A watch function, which may be sync or async.
                 old_value: The old value of the attribute.
                 value: The new value of the attribute.
@@ -239,17 +243,15 @@ class Reactive(Generic[ReactiveType]):
                 watch_result = watch_function()
             if isawaitable(watch_result):
                 # Result is awaitable, so we need to await it within an async context
-                obj.post_message(
-                    events.Callback(callback=partial(await_watcher, watch_result))
-                )
+                watcher_object.call_next(partial(await_watcher, watch_result))
 
         private_watch_function = getattr(obj, f"_watch_{name}", None)
         if callable(private_watch_function):
-            invoke_watcher(private_watch_function, old_value, value)
+            invoke_watcher(obj, private_watch_function, old_value, value)
 
         public_watch_function = getattr(obj, f"watch_{name}", None)
         if callable(public_watch_function):
-            invoke_watcher(public_watch_function, old_value, value)
+            invoke_watcher(obj, public_watch_function, old_value, value)
 
         # Process "global" watchers
         watchers: list[tuple[Reactable, Callable]]
@@ -263,7 +265,7 @@ class Reactive(Generic[ReactiveType]):
             ]
             for reactable, callback in watchers:
                 with reactable.prevent(*obj._prevent_message_types_stack[-1]):
-                    invoke_watcher(callback, old_value, value)
+                    invoke_watcher(reactable, callback, old_value, value)
 
     @classmethod
     def _compute(cls, obj: Reactable) -> None:
diff --git a/tests/test_call_later.py b/tests/test_call_x_schedulers.py
similarity index 100%
rename from tests/test_call_later.py
rename to tests/test_call_x_schedulers.py
diff --git a/tests/test_message_pump.py b/tests/test_message_pump.py
index c02978e69..c6f9d921c 100644
--- a/tests/test_message_pump.py
+++ b/tests/test_message_pump.py
@@ -87,3 +87,39 @@ async def test_prevent() -> None:
         await pilot.pause()
         assert len(app.input_changed_events) == 1
         assert app.input_changed_events[0].value == "foo"
+
+
+async def test_prevent_with_call_next() -> None:
+    """Test for https://github.com/Textualize/textual/issues/3166.
+
+    Does a callback scheduled with `call_next` respect messages that
+    were prevented when it was scheduled?
+    """
+
+    hits = 0
+
+    class PreventTestApp(App[None]):
+        def compose(self) -> ComposeResult:
+            yield Input()
+
+        def change_input(self) -> None:
+            self.query_one(Input).value += "a"
+
+        def on_input_changed(self) -> None:
+            nonlocal hits
+            hits += 1
+
+    app = PreventTestApp()
+    async with app.run_test() as pilot:
+        app.call_next(app.change_input)
+        await pilot.pause()
+        assert hits == 1
+
+        with app.prevent(Input.Changed):
+            app.call_next(app.change_input)
+        await pilot.pause()
+        assert hits == 1
+
+        app.call_next(app.change_input)
+        await pilot.pause()
+        assert hits == 2
diff --git a/tests/test_reactive.py b/tests/test_reactive.py
index cb5a6b5f2..9ab1af192 100644
--- a/tests/test_reactive.py
+++ b/tests/test_reactive.py
@@ -499,3 +499,81 @@ async def test_private_compute() -> None:
     async with PrivateComputeTest().run_test() as pilot:
         pilot.app.base = 5
         assert pilot.app.double == 10
+
+
+async def test_async_reactive_watch_callbacks_go_on_the_watcher():
+    """Regression test for https://github.com/Textualize/textual/issues/3036.
+
+    This makes sure that async callbacks are called.
+    See the next test for sync callbacks.
+    """
+
+    from_app = False
+    from_holder = False
+
+    class Holder(Widget):
+        attr = var(None)
+
+        def watch_attr(self):
+            nonlocal from_holder
+            from_holder = True
+
+    class MyApp(App):
+        def __init__(self):
+            super().__init__()
+            self.holder = Holder()
+
+        def on_mount(self):
+            self.watch(self.holder, "attr", self.callback)
+
+        def update(self):
+            self.holder.attr = "hello world"
+
+        async def callback(self):
+            nonlocal from_app
+            from_app = True
+
+    async with MyApp().run_test() as pilot:
+        pilot.app.update()
+        await pilot.pause()
+        assert from_holder
+        assert from_app
+
+
+async def test_sync_reactive_watch_callbacks_go_on_the_watcher():
+    """Regression test for https://github.com/Textualize/textual/issues/3036.
+
+    This makes sure that sync callbacks are called.
+    See the previous test for async callbacks.
+    """
+
+    from_app = False
+    from_holder = False
+
+    class Holder(Widget):
+        attr = var(None)
+
+        def watch_attr(self):
+            nonlocal from_holder
+            from_holder = True
+
+    class MyApp(App):
+        def __init__(self):
+            super().__init__()
+            self.holder = Holder()
+
+        def on_mount(self):
+            self.watch(self.holder, "attr", self.callback)
+
+        def update(self):
+            self.holder.attr = "hello world"
+
+        def callback(self):
+            nonlocal from_app
+            from_app = True
+
+    async with MyApp().run_test() as pilot:
+        pilot.app.update()
+        await pilot.pause()
+        assert from_holder
+        assert from_app