Close unawaited watcher coroutines when messages are dropped

evidence/reactive-callback-retirement-20261007/README.md

@@ -0,0 +1,45 @@
+# Deferred watcher ownership
+
+Async watcher results were created by `invoke_watcher`, then hidden in a
+`partial(await_watcher, ...)`. Dropping an undelivered callback did not close
+the coroutine. A receiver can also be constructed and abandoned before its
+message task ever starts; task exit alone cannot cover that lifetime.
+
+`WatcherCallback`, an original Callback subtype, now owns that result's await
+and disposal. `Message._discard` supplies the delivery resource hook.
+MessagePump releases rejected messages, abandoned callback batches, and pending
+messages on task exit or never-started receiver destruction. Actual watcher
+invocation stays immediate, arguments retain their original values, and compute
+still follows successful completion. Disposing a delivery closes native
+coroutines; it does not cancel independently owned Tasks or Futures.
+
+The source pass covered Reactive invocation/reset/subscriptions, Callback
+dispatch, Message completion/bubbling/replacement, pump admission/startup/close/
+batch cancellation, and the call_next consumers in App, Widget, Screen, Signal,
+Lazy, AwaitMount, AwaitRemove and AwaitComplete. No dispatch supply, directory
+watcher, Toad caller, or frame publication behavior was replaced.
+The existing Package AST reader parsed 250 native modules with no omissions;
+the source search found no competing native or current Toad `__del__` owner.
+
+## Verification
+
+- Existing reactive and message-pump checks plus five lifetime checks:
+  **46 passed, 1.19s**. These cover immediate synchronous invocation, borrowed
+  Task ownership, running/pending watcher cancellation, closed receiver
+  rejection, never-mounted DirectoryTree disposal, and real tree removal.
+- Original `tests/sidebar_retirement_pilot.py`, unchanged, using the released
+  sidebar-eviction author runtime with this native source: **exit 0, empty
+  stderr**. All three cycles restored actual selected filenames and scroll
+  positions, collected retired panel graphs, and completed source replacement
+  and hydration cancellation assertions. No provider inputs or package changes.
+
+Logs: `/home/ts/.cache/agent-scratch/native-watcher-retirement-20261007/`.
+`native-final.log`, `app02.stdout.log`, and `app02.stderr.log` are the final
+results. `app.stdout.log` / `app.stderr.log` preserve the initial App negative:
+restoration passed but the 161-byte unawaited-coroutine warning remained before
+never-started receiver disposal was added. An attempted callback-only finalizer
+also failed the unmounted-tree check; that competing finalizer was deleted.
+
+This is native source plus affected source-App acceptance. No installed wheel,
+live UI, or performance improvement is claimed. The original app fixture owns
+its private child cleanup; exact child birth records were not collected here.

src/textual/message.py

@@ -62,6 +62,9 @@ def __post_init__(self) -> None:
     def __rich_repr__(self) -> rich.repr.Result:
         yield from ()
 
+    def _discard(self) -> None:
+        """Release resources owned by an abandoned or completed delivery."""
+
     def __init_subclass__(
         cls,
         bubble: bool | None = True,

src/textual/message_pump.py

@@ -552,7 +552,9 @@ def call_later(self, callback: Callback, *args: Any, **kwargs: Any) -> bool:
         message = events.Callback(callback=partial(callback, *args, **kwargs))
         return self.post_message(message)
 
-    def call_next(self, callback: Callback, *args: Any, **kwargs: Any) -> None:
+    def call_next(
+        self, callback: Callback | events.Callback, *args: Any, **kwargs: Any
+    ) -> None:
         """Schedule a callback to run immediately after processing the current message.
 
         Args:
@@ -561,11 +563,38 @@ def call_next(self, callback: Callback, *args: Any, **kwargs: Any) -> None:
             **kwargs: Keyword arguments to pass to the callable.
         """
         assert callback is not None, "Callback must not be None"
-        callback_message = events.Callback(callback=partial(callback, *args, **kwargs))
+        if isinstance(callback, events.Callback):
+            if args or kwargs:
+                raise TypeError("A callback event already owns its arguments")
+            callback_message = callback
+        else:
+            callback_message = events.Callback(callback=partial(callback, *args, **kwargs))
+        if self._closing or self._closed:
+            callback_message._discard()
+            return
         callback_message._prevent.update(self._get_prevented_messages())
         self._next_callbacks.append(callback_message)
         self.check_idle()
 
+    def _discard_next_callbacks(self) -> None:
+        for callback in self.__dict__.pop("_next_callbacks", ()):
+            callback._discard()
+
+    def _discard_pending_messages(self) -> None:
+        """Release undelivered messages after their receiver has stopped."""
+        self._discard_next_callbacks()
+        queue = self.__dict__.get("_message_queue")
+        if queue is not None:
+            while not queue.empty():
+                message = queue.get_nowait()
+                if message is not None:
+                    message._discard()
+
+    def __del__(self) -> None:
+        # Constructor watchers may enqueue work before a widget is mounted.
+        # An abandoned, never-started receiver still owns those deliveries.
+        self._discard_pending_messages()
+
     def _on_invoke_later(self, message: messages.InvokeLater) -> None:
         # Forward InvokeLater message to the Screen
         if self.app._running:
@@ -582,6 +611,7 @@ async def _close_messages(self, wait: bool = True) -> None:
             return
         self._closing = True
         self._message_pump_closing()
+        self._discard_next_callbacks()
         self._clear_signal_subscriptions()
         Reactive._clear_watch_subscriptions(self)
         if self._timers:
@@ -617,6 +647,7 @@ def _start_messages(self) -> None:
             self._closing = True
             self._closed = True
             self._message_pump_closing()
+            self._discard_pending_messages()
 
     async def _process_messages(self, **kwargs: Any) -> None:
         """Own the real pump task through subclass startup, dispatch and exit."""
@@ -629,6 +660,7 @@ async def _process_messages(self, **kwargs: Any) -> None:
             # node despite its weak subscriber key. Teardown owns their removal,
             # including cancellation and failed pre-processing.
             self._running = False
+            self._discard_pending_messages()
             self._clear_signal_subscriptions()
             Reactive._clear_watch_subscriptions(self)
             self._task = None
@@ -745,6 +777,7 @@ async def _process_messages_loop(self) -> None:
                 break
             finally:
                 message._complete_dispatch(self, result)
+                message._discard()
                 # Subscription access initializes this optional signal. Do not
                 # allocate its weak-key subscriber machinery for quiet pumps.
                 message_signal = self.__dict__.get("message_signal")
@@ -787,13 +820,17 @@ async def _flush_next_callbacks(self) -> None:
         """Invoke pending callbacks in next callbacks queue."""
         callbacks = self._next_callbacks.copy()
         self._next_callbacks.clear()
-        for callback in callbacks:
-            try:
-                with self.prevent(*callback._prevent):
-                    await invoke(callback.callback)
-            except Exception as error:
-                self.app._handle_exception(error)
-                break
+        try:
+            for callback in callbacks:
+                try:
+                    with self.prevent(*callback._prevent):
+                        await invoke(callback.callback)
+                except Exception as error:
+                    self.app._handle_exception(error)
+                    break
+        finally:
+            for callback in callbacks:
+                callback._discard()
 
     async def _dispatch_message(self, message: Message) -> Message | None:
         """Dispatch a message received from the message queue.
@@ -1006,8 +1043,10 @@ def post_message(self, message: Message) -> bool:
                 "Message is missing attributes; did you forget to call super().__init__() ?"
             )
         if self._closing or self._closed:
+            message._discard()
             return False
         if not self.check_message_enabled(message):
+            message._discard()
             return False
         # Add a copy of the prevented message types to the message
         # This is so that prevented messages are honoured by the event's handler

src/textual/reactive.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from functools import partial
-from inspect import isawaitable
+from inspect import isawaitable, iscoroutine
 from operator import methodcaller
 from weakref import WeakKeyDictionary
 from typing import (
@@ -81,12 +81,31 @@ def __call__(self, obj: ReactableType) -> ReactiveType:
         return self.callback(obj)
 
 
-async def await_watcher(obj: Reactable, awaitable: Awaitable[object]) -> None:
-    """Coroutine to await an awaitable returned from a watcher"""
-    _rich_traceback_omit = True
-    await awaitable
-    # Watcher may have changed the state, so run compute again
-    obj.post_message(events.Callback(callback=partial(Reactive._compute, obj)))
+class WatcherCallback(events.Callback):
+    """Own a watcher's already-created result through deferred delivery."""
+
+    def __init__(self, obj: Reactable, awaitable: Awaitable[object]) -> None:
+        self.obj = obj
+        self.awaitable = awaitable
+        super().__init__(self._await_result)
+
+    async def _await_result(self) -> None:
+        await self.awaitable
+        # Watcher may have changed the state, so run compute again.
+        self.obj.post_message(
+            events.Callback(callback=partial(Reactive._compute, self.obj))
+        )
+
+    def _discard(self) -> None:
+        # A freshly returned coroutine belongs to this deferred delivery.
+        # Tasks / Futures have an independent execution owner; do not cancel
+        # them merely because this callback was never admitted.
+        if iscoroutine(self.awaitable):
+            self.awaitable.close()
+
+    def __rich_repr__(self) -> rich.repr.Result:
+        yield "obj", self.obj
+        yield "awaitable", self.awaitable
 
 
 class _StoredReactiveAccess:
@@ -172,9 +191,7 @@ def invoke_watcher(
             watch_result = cast(WatchCallbackNoArgsType, watch_function)()
         if isawaitable(watch_result):
             # Result is awaitable, so we need to await it within an async context
-            watcher_object.call_next(
-                partial(await_watcher, watcher_object, watch_result)
-            )
+            watcher_object.call_next(WatcherCallback(watcher_object, watch_result))
 
 
 @rich.repr.auto

tests/test_reactive.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import asyncio
+import gc
+from inspect import CORO_CLOSED, getcoroutinestate
 
 import pytest
 
@@ -9,11 +11,127 @@
 from textual.message_pump import MessagePump
 from textual.reactive import Initialize, Reactive, TooManyComputesError, reactive, var
 from textual.widget import Widget
+from textual.widgets import DirectoryTree
 
 OLD_VALUE = 5_000
 NEW_VALUE = 1_000_000
 
 
+async def test_abandoned_watcher_results_keep_invocation_and_borrowed_task_lifetime():
+    """Retirement closes owned coroutines without cancelling borrowed execution."""
+    results = []
+    calls = []
+    release = asyncio.Event()
+
+    async def work():
+        await release.wait()
+
+    class WatchWidget(Widget):
+        count = var(0, init=False)
+
+        def watch_count(self, value):
+            calls.append(value)
+            result = work() if value == 1 else asyncio.create_task(work())
+            results.append(result)
+            return result
+
+    async with App().run_test():
+        widget = WatchWidget()
+        widget.count = 1
+        widget.count = 2
+        assert calls == [1, 2]
+        await widget._close_messages(wait=False)
+        assert getcoroutinestate(results[0]) == CORO_CLOSED
+        assert not results[1].cancelled()
+        release.set()
+        await results[1]
+
+
+async def test_cancelled_watcher_batch_closes_active_and_unstarted_results():
+    started = asyncio.Event()
+    results = []
+
+    async def work():
+        started.set()
+        await asyncio.Event().wait()
+
+    class WatchWidget(Widget):
+        count = var(0, init=False)
+
+        def watch_count(self):
+            result = work()
+            results.append(result)
+            return result
+
+    async with App().run_test():
+        widget = WatchWidget()
+        widget.count = 1
+        widget.count = 2
+        batch = asyncio.create_task(widget._flush_next_callbacks())
+        await started.wait()
+        batch.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await batch
+        assert [getcoroutinestate(result) for result in results] == [
+            CORO_CLOSED,
+            CORO_CLOSED,
+        ]
+
+
+async def test_watcher_created_after_retirement_is_disposed():
+    results = []
+
+    async def work():
+        raise AssertionError("A closed receiver must not run its watcher")
+
+    class WatchWidget(Widget):
+        count = var(0, init=False)
+
+        def watch_count(self):
+            result = work()
+            results.append(result)
+            return result
+
+    async with App().run_test():
+        widget = WatchWidget()
+        await widget._close_messages(wait=False)
+        widget.count = 1
+        assert getcoroutinestate(results[0]) == CORO_CLOSED
+
+
+async def test_unmounted_directory_tree_releases_initial_watcher(tmp_path, recwarn):
+    async with App().run_test():
+        tree = DirectoryTree(tmp_path)
+        del tree
+        gc.collect()
+    assert not [
+        warning for warning in recwarn if "was never awaited" in str(warning.message)
+    ]
+
+
+async def test_directory_tree_path_changes_and_removal_release_watchers(
+    tmp_path, recwarn
+):
+    """Exercise the real directory watcher and removal, including pending delivery."""
+    first = tmp_path / "first"
+    second = tmp_path / "second"
+    first.mkdir()
+    second.mkdir()
+    (first / "one.txt").write_text("one")
+    (second / "two.txt").write_text("two")
+
+    async with App().run_test() as pilot:
+        for _ in range(3):
+            tree = DirectoryTree(first)
+            await pilot.app.mount(tree)
+            tree.path = second
+            await tree.remove()
+    gc.collect()
+    assert not [
+        warning for warning in recwarn if "was never awaited" in str(warning.message)
+    ]
+
+
 async def test_initialize():
     """Test that the default accepts an Initialize instance."""